Flink 从 kafka 中读取数据并输出到 kafka

Kafka 是一个分布式的基于发布/订阅的消息系统，本身处理的也是流式数据。kafka和flink二者被称为当前处理流式数据的双子星。

下面我们将从以下几个步骤展开讲解：

一、添加maven依赖

<!--kafka connector-->
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-kafka_2.12</artifactId>
    <version>1.13.1</version>
</dependency>

二、编写flink程序

老规矩，先上代码 再做介绍

代码如下：

package com.flink.wc.myflink.source;

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.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.util.Properties;

public class mysource_kafka_kafka {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 配置kafka集群信息  properties是java中的一个集合类, 多用于 配置参数, 它继承于 Hashtable，表示一个持久的属性集.属性列表中每个键及其对应值都是一个字符串。
        // 这里和在kafka javaAPI中配置kafka信息时一样
        Properties properties = new Properties();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop-001:9092");
            // 配置序列化 
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
            // 配置消费者组
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "test");
        
        // 从kafka中读取数据
        DataStreamSource<String> stream = env.addSource(new FlinkKafkaConsumer<String>("myflink_source", new SimpleStringSchema(), properties));

        stream.print("flink");

        // 将数据输出到kafka
        stream.addSink(new FlinkKafkaProducer<String>("myflink_sink", new SimpleStringSchema(), properties));
        
        env.execute();

    }
}

1、从kafka读取数据

通过addSource()方法传入一个SourceFunction的实现类

FlinkKafkaConsumer（） 就是这个实现类 很好理解 就是实例化一个flink程序的kafka消费者

源代码中FlinkKafkaConsumer类 构造函数如下：三个参数分别是 （kafka主题, 反序列化对象, kafka集群配置信息）

public FlinkKafkaConsumer(String topic, DeserializationSchema<T> valueDeserializer, Properties props) {
    this(Collections.singletonList(topic), valueDeserializer, props);
}

2、输出数据到kakfka

通过addSink()方法传入一个SinkFunction的实现类

FlinkKafkaProducer （） 就是这个实现类 也很好理解 就是实例化一个flink程序的kafka生产者

构造函数如下：三个参数分别是 （kafka主题, 序列化对象, kafka集群配置信息）

public FlinkKafkaProducer(String topicId, SerializationSchema<IN> serializationSchema, Properties producerConfig) {
    this(topicId, serializationSchema, producerConfig, Optional.of(new FlinkFixedPartitioner()));
}

三、启动kafka集群

// 启动zookeeper
// 启动kafka
(base) [hadoop@hadoop-001 ~]$ jps
1410 QuorumPeerMain
6583 NameNode
8121 Jps
8058 Kafka
6798 DataNode

// hadoop-001 上启动生产者:
(base) [hadoop@hadoop-001 ~]$ kafka-console-producer.sh --bootstrap-server hadoop-001:9092 --topic myflink_source

// hadoop-002 上启动消费者 :
(base) [hadoop@hadoop-002 ~]$ kafka-console-consumer.sh --bootstrap-server hadoop-001:9092 --topic myflink_sink

四、运行flink程序