Flink 1.17教程：输出算子之输出到Kafka

淡淡的烟草味﹌ 2023-10-15 18:20 50阅读 0赞

### 输出到Kafka ###

（1）添加Kafka 连接器依赖  
由于我们已经测试过从Kafka数据源读取数据，连接器相关依赖已经引入，这里就不重复介绍了。  
（2）启动Kafka集群  
（3）编写输出到Kafka的示例代码  
输出无key的record:

public class SinkKafka {
        
        public static void main(String[] args) throws Exception {
        
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setParallelism(1);
    
            // 如果是精准一次，必须开启checkpoint（后续章节介绍）
            env.enableCheckpointing(2000, CheckpointingMode.EXACTLY_ONCE);
    
    
            SingleOutputStreamOperator<String> sensorDS = env
                    .socketTextStream("hadoop102", 7777);
    
            /**
             * Kafka Sink:
             * TODO 注意：如果要使用 精准一次 写入Kafka，需要满足以下条件，缺一不可
             * 1、开启checkpoint（后续介绍）
             * 2、设置事务前缀
             * 3、设置事务超时时间：   checkpoint间隔 <  事务超时时间  < max的15分钟
             */
            KafkaSink<String> kafkaSink = KafkaSink.<String>builder()
                    // 指定 kafka 的地址和端口
                    .setBootstrapServers("hadoop102:9092,hadoop103:9092,hadoop104:9092")
                    // 指定序列化器：指定Topic名称、具体的序列化
                    .setRecordSerializer(
                            KafkaRecordSerializationSchema.<String>builder()
                                    .setTopic("ws")
                                    .setValueSerializationSchema(new SimpleStringSchema())
                                    .build()
                    )
                    // 写到kafka的一致性级别： 精准一次、至少一次
    				.setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
                    // 如果是精准一次，必须设置 事务的前缀
                    .setTransactionalIdPrefix("atguigu-")
                    // 如果是精准一次，必须设置 事务超时时间: 大于checkpoint间隔，小于 max 15分钟
                    .setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, 10*60*1000+"")
                    .build();
    
    
            sensorDS.sinkTo(kafkaSink);
    
    
            env.execute();
        }
    }

自定义序列化器，实现带key的record:

public class SinkKafkaWithKey {
        
        public static void main(String[] args) throws Exception {
        
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setParallelism(1);
    
            env.enableCheckpointing(2000, CheckpointingMode.EXACTLY_ONCE);
            env.setRestartStrategy(RestartStrategies.noRestart());
    
    
            SingleOutputStreamOperator<String> sensorDS = env
                    .socketTextStream("hadoop102", 7777);
    
    
            /**
             * 如果要指定写入kafka的key，可以自定义序列化器：
             * 1、实现 一个接口，重写 序列化 方法
             * 2、指定key，转成 字节数组
             * 3、指定value，转成 字节数组
             * 4、返回一个 ProducerRecord对象，把key、value放进去
             */
            KafkaSink<String> kafkaSink = KafkaSink.<String>builder()
                    .setBootstrapServers("hadoop102:9092,hadoop103:9092,hadoop104:9092")
                    .setRecordSerializer(
                            new KafkaRecordSerializationSchema<String>() {
        
    
                                @Nullable
                                @Override
                                public ProducerRecord<byte[], byte[]> serialize(String element, KafkaSinkContext context, Long timestamp) {
        
                                    String[] datas = element.split(",");
                                    byte[] key = datas[0].getBytes(StandardCharsets.UTF_8);
                                    byte[] value = element.getBytes(StandardCharsets.UTF_8);
                                    return new ProducerRecord<>("ws", key, value);
                                }
                            }
                    )
                    .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
                    .setTransactionalIdPrefix("atguigu-")
                    .setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, 10 * 60 * 1000 + "")
                    .build();
    
    
            sensorDS.sinkTo(kafkaSink);
    
    
            env.execute();
        }
    }

（4）运行代码，在Linux主机启动一个消费者，查看是否收到数据

[atguigu@hadoop102 ~]$ 
    bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic ws