【Flink1.14实战】Docker环境 DataStream kafka Sink
DataStream 连接器
要在应用程序中使用这些连接器之一，通常需要额外的第三方组件，例如用于数据存储或消息队列的服务器。另请注意，虽然本节中列出的流连接器是 Flink 项目的一部分并且包含在源代码版本中，但它们不包含在二进制发行版中。
kafka 连接器
该文档描述的是基于新数据源 API的 Kafka Source。
依赖
Apache Flink 集成了通用的 Kafka 连接器，它会尽力与 Kafka client 的最新版本保持同步。该连接器使用的 Kafka client 版本可能会在 Flink 版本之间发生变化。 当前 Kafka client 向后兼容 0.10.0 或更高版本的 Kafka broker。
<dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-connector-kafka_2.11</artifactId> <version>1.14.4</version></dependency>Flink 目前的流连接器还不是二进制发行版的一部分。
Kafka Sink
KafkaSink 可将数据流写入一个或多个 Kafka topic。
使用方法
Kafka sink 提供了构建类来创建 KafkaSink 的实例。以下代码片段展示了如何将字符串数据按照至少一次（at lease once）的语义保证写入 Kafka topic：
DataStream<String> stream = ...; KafkaSink<String> sink = KafkaSink.<String>builder() .setBootstrapServers(brokers) .setRecordSerializer(KafkaRecordSerializationSchema.builder() .setTopic("topic-name") .setValueSerializationSchema(new SimpleStringSchema()) .build() ) .setDeliverGuarantee(DeliveryGuarantee.AT_LEAST_ONCE) .build(); stream.sinkTo(sink);以下属性在构建 KafkaSink 时是必须指定的：
- Bootstrap servers,
setBootstrapServers(String) - 消息序列化器（Serializer）,
setRecordSerializer(KafkaRecordSerializationSchema) - 如果使用
DeliveryGuarantee.EXACTLY_ONCE的语义保证，则需要使用setTransactionalIdPrefix(String)
实战
1、编写kafka数据源程序
package quick;import org.apache.flink.api.common.eventtime.WatermarkStrategy;import org.apache.flink.api.common.functions.FlatMapFunction;import org.apache.flink.api.common.serialization.SimpleStringSchema;import org.apache.flink.api.java.tuple.Tuple2;import org.apache.flink.api.java.utils.ParameterTool;import org.apache.flink.connector.base.DeliveryGuarantee;import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;import org.apache.flink.connector.kafka.sink.KafkaSink;import org.apache.flink.connector.kafka.source.KafkaSource;import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;import org.apache.flink.streaming.api.datastream.DataStream;import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;import org.apache.flink.streaming.api.windowing.time.Time;import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;import org.apache.flink.util.Collector;import quick.clickcount.records.ClickEventStatisticsSerializationSchema;public class KafkaSinkExample { public static void main(String[] args) throws Exception{ final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); final ParameterTool params = ParameterTool.fromArgs(args); String brokers = params.get("bootstrap.servers", "kafka:9092"); KafkaSource source = KafkaSource.builder() .setBootstrapServers(brokers) .setTopics("input") .setGroupId("my-group") .setStartingOffsets(OffsetsInitializer.earliest()) .setValueOnlyDeserializer(new SimpleStringSchema()) .build(); DataStream dataStream = env .fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source") .flatMap(new Splitter()); KafkaSink sink = KafkaSink.builder() .setBootstrapServers(brokers) .setRecordSerializer(KafkaRecordSerializationSchema.builder() .setTopic("output") .setValueSerializationSchema(new SimpleStringSchema()) .build() ) .setDeliverGuarantee(DeliveryGuarantee.AT_LEAST_ONCE) .build(); dataStream.sinkTo(sink); env.execute("KafkaSinkExample job"); } public static class Splitter implements FlatMapFunction { @Override public void flatMap(String sentence, Collector out) throws Exception { for (String word: sentence.split(" ")) { out.collect(word); } } }}2、flink日志查看
然后，将打包应用程序提交，Flink 的Web UI来提交作业监控集群的状态和正在运行的作业。
只需输入一些单词，然后按回车键即可传入新单词。这些将作为单词统计程序的输入。如果想查看大于 1 的计数，在 5 秒内重复输入相同的单词即可（如果无法快速输入，则可以将窗口大小从 5 秒增加 ☺）。
$ docker-compose logs -f taskmanager3、kafka生产者不断提交数据
$ docker exec -it kafka_kafka_1 /bin/bash $ cd /opt/kafka$ bin/kafka-console-producer.sh --broker-list localhost:9092 --topic inputThis is a messageThis is another message4、另外打开一终端，kafka消费者查看处理的数据，即将输入句子分割成单词
$ docker exec -it kafka_kafka_1 /bin/bash $ cd /opt/kafka$ bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic output --from-beginningThisisanothermessage