说明

做数据分析时，有时需要将初步统计的数据，发送到 kafka中，以便进一步处理

Kafka 生产者连接器 api

public FlinkKafkaProducer(
  String topicId,
  SerializationSchema<IN> serializationSchema,
  Properties producerConfig
)

形参解释：

• topicId：主题
• serializationSchema：序列化器
• producerConfig：
  • kafka服务器ip、port

案例

pom.xml

需要添加依赖：

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-kafka_2.12</artifactId>
    <version>${flink.version}</version>
</dependency>

java

package sink;

import org.apache.flink.api.common.serialization.SerializationSchema;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.kafka.clients.consumer.ProducerConfig;

import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

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

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

        DataStreamSource<String> dss = env.fromElements("lilei", "lili", "lilei", "lucy", "lili");

        SingleOutputStreamOperator<Tuple2<String, Integer>> mapSOSO = dss.map(item -> Tuple2.of(item, 1))
                .returns(Types.TUPLE(Types.STRING,Types.INT));

        SingleOutputStreamOperator<Tuple2<String, Integer>> sumSOSO = mapSOSO.keyBy(item -> item.f0).sum(1);

        Properties properties = new Properties();
        properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop1:9092,hadoop2:9092,hadoop3:9092");

        FlinkKafkaProducer<Tuple2<String, Integer>> myProducer = new FlinkKafkaProducer(
                "flink2kafka", // 主题 topic,
                new JsonSerializationSchema(),
                properties); // 容错

        sumSOSO.addSink(myProducer);
        sumSOSO.print();
        env.execute();
    }

    // 定义序列化器
    static class JsonSerializationSchema implements SerializationSchema<Tuple2<String, Integer>>{

        @Override
        public byte[] serialize(Tuple2<String, Integer> element) {
            ObjectMapper om = new ObjectMapper();
            Map map = new HashMap();
            byte[] arr = null;
            try {
                // key是 tuple的第一个元素，即：人名
                // value是 tuple的第二个元素，即：人名出现的次数
                map.put(element.f0,element.f1);
                arr = om.writeValueAsBytes(map);
            } catch (Exception e) {
                e.printStackTrace();
            }
            return arr;
        }
    }
}

启动zk

登录 hadoop1，执行下面命令：

/program/bin/zk.sh start

启动kafka

登录 hadoop1，执行下面命令：

/program/bin/kafka.sh start

启动 flink

在 idea 中运行 flink 程序

启动消费者、指定主题

登录 hadoop1，执行下面命令：

/program/kafka_2.12-2.4.1/bin/kafka-console-consumer.sh --bootstrap-server hadoop1:9092,hadoop2:9092,hadoop3:9092 --topic flink2kafka

提示 如果先启动消费者，并且指定 新主题，会报错，错误如下：

Error while fetching metadata with correlation id 2

原文出处：/show_1IX45IVVX9Om.html