flink将数据写入到kafka

flink将数据写入到kafka 前言

在之前的一篇中，我们谈到在flink中如何读取来自kafka指定topic的数据，在实际生产环境中，比较常见的场景是，kafka作为数据源，flink程序读取到kafka的数据后，进行初步的处理后，再将数据会写到kafka的另一个topic中，供后面的应用读取使用

本例模拟从外部文件中读取数据，然后写入到kafka的指定topic中

环境准备

启动kafka服务，并通过控制台启动kafka指定topic的消费端窗口，这里我们还是使用之前那个 "zcy"的topic，使用下面的命令启动消费端窗口

./kafka-console-consumer.sh --bootstrap-server IP:9092 --topic zcy

我们仍然读取下面的文件数据

 

核心代码编写

import com.congge.source.SensorReading;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer011;

public class SinkKafka {

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

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //从环境的集合中获取数据
        String path = "E:\code-self\flink_study\src\main\resources\sensor.txt";
        DataStreamSource inputStream = env.readTextFile(path);
        DataStream resultMesage = inputStream.map(new MapFunction() {
            @Override
            public String map(String value) throws Exception {
                String[] fields = value.split(",");
                return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2])).toString();
            }
        });

        resultMesage.addSink(new FlinkKafkaProducer011("IP:9092","zcy",new SimpleStringSchema()));
        env.execute();
    }

}

 运行上面的程序，可以看到 zcy 这个topic下成功消费到了数据