[大数据]Flink从Kafka读取数据

Flink从Kafka读取数据

1、环境搭建，hadoop集群、zookeeper、kafka

2、IDEA程序编写+配置依赖

在IDEA中添加kafka配置依赖：

Apache Flink 附带了一个通用的 Kafka 连接器，它试图跟踪最新版本的 Kafka 客户端。它使用的客户端版本可能会在 Flink 版本之间发生变化。现代 Kafka 客户端向后兼容代理版本 0.10.0 或更高版本。关于Kafka兼容性的详细信息，请参考Kafka官方文档。

<dependency>
	<groupId>org.apache.flink</groupId>
	<artifactId>flink-connector-kafka_2.11</artifactId>
	<version>1.12.3</version>
</dependency>

IDEA程序编写：

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;

import java.util.Properties;


public class Flink_Source_Kafka {
    public static void main(String[] args) throws Exception {
        //1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //2.从kafka读取数据
        Properties properties = new Properties();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop102:9092");
        properties.put(ConsumerConfig.GROUP_ID_CONFIG,"comexercise");
        DataStreamSource<String> kafkaDS = env.addSource(new FlinkKafkaConsumer<String>("test", new SimpleStringSchema(), properties));
        
        //3.将数据打印
        kafkaDS.print();
        //4.执行任务
        env.execute();

    }
}

3、在kafka中创建一个生产者（producer）

在kafka的bin文件下输入命令：

kafka-console-producer.sh --broker-list hadoop102:9092 --topic test;

4、启动idea程序，在kafka生产者中输入数据

5、kafka常用命令：

1、查看所有的话题

./kafka-topics.sh --list --zookeeper localhost:9092

2、查看所有话题的详细信息

./kafka-topics.sh --zookeeper localhost:2181 --describe

3、列出指定话题的详细信息

./kafka-topics.sh --zookeeper localhost:2181 --describe  --topic demo

4、删除一个话题

./kafka-topics.sh --zookeeper localhost:2181 --delete  --topic test

5、创建一个叫test的话题，有两个分区，每个分区3个副本

./kafka-topics.sh --zookeeper localhost:2181 --create --topic test --replication-factor 3 --partitions 2

6、测试kafka发送和接收消息（启动两个终端）

#发送消息（注意端口号为配置文件里面的端口号）
./kafka-console-producer.sh --broker-list localhost:9092 --topic test
#消费消息（可能端口号与配置文件保持一致，或与发送端口保持一致）
./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning   #加了--from-beginning 重头消费所有的消息
./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test         #不加--from-beginning 从最新的一条消息开始消费

7、查看某个topic对应的消息数量

./kafka-run-class.sh  kafka.tools.GetOffsetShell --broker-list localhost:9092 --topic test --time -1

8、显示所有消费者

./kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list

9、获取正在消费的topic（console-consumer-63307）的group的offset

./kafka-consumer-groups.sh --describe --group console-consumer-63307 --bootstrap-server localhost:9092

10、显示消费者

./kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list
--bootstrap-server localhost:9092

10、显示消费者

./kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list