《从0到1学习Flink》—— Flink 写入数据到 Kafka

时间 2020-01-28

标签从0到1学习Flink flink 写入数据 kafka 栏目 Kafka 繁體版

原文原文链接

前言

以前文章《从0到1学习Flink》—— Flink 写入数据到 ElasticSearch 写了如何将 Kafka 中的数据存储到 ElasticSearch 中，里面其实就已经用到了 Flink 自带的 Kafka source connector（FlinkKafkaConsumer）。存入到 ES 只是其中一种状况，那么若是咱们有多个地方须要这份经过 Flink 转换后的数据，是否是又要咱们继续写个 sink 的插件呢？确实，因此 Flink 里面就默认支持了很多 sink，好比也支持 Kafka sink connector（FlinkKafkaProducer），那么这篇文章咱们就讲讲如何将数据写入到 Kafka。java

准备

添加依赖

Flink 里面支持 Kafka 0.八、0.九、0.十、0.11 ，之后有时间能够分析下源码的实现。git

这里咱们须要安装下 Kafka，请对应添加对应的 Flink Kafka connector 依赖的版本，这里咱们使用的是 0.11 版本：github

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-kafka-0.11_2.11</artifactId>
    <version>${flink.version}</version>
</dependency>

Kafka 安装

这里就不写这块内容了，能够参考我之前的文章 Kafka 安装及快速入门。apache

这里咱们演示把其余 Kafka 集群中 topic 数据原样写入到本身本地起的 Kafka 中去。微信

配置文件

kafka.brokers=xxx:9092,xxx:9092,xxx:9092
kafka.group.id=metrics-group-test
kafka.zookeeper.connect=xxx:2181
metrics.topic=xxx
stream.parallelism=5
kafka.sink.brokers=localhost:9092
kafka.sink.topic=metric-test
stream.checkpoint.interval=1000
stream.checkpoint.enable=false
stream.sink.parallelism=5

目前咱们先看下本地 Kafka 是否有这个 metric-test topic 呢？须要执行下这个命令：学习

bin/kafka-topics.sh --list --zookeeper localhost:2181

能够看到本地的 Kafka 是没有任何 topic 的，若是等下咱们的程序运行起来后，再次执行这个命令出现 metric-test topic，那么证实个人程序确实起做用了，已经将其余集群的 Kafka 数据写入到本地 Kafka 了。ui

程序代码

Main.javaspa

public class Main {
    public static void main(String[] args) throws Exception{
        final ParameterTool parameterTool = ExecutionEnvUtil.createParameterTool(args);
        StreamExecutionEnvironment env = ExecutionEnvUtil.prepare(parameterTool);
        DataStreamSource<Metrics> data = KafkaConfigUtil.buildSource(env);

        data.addSink(new FlinkKafkaProducer011<Metrics>(
                parameterTool.get("kafka.sink.brokers"),
                parameterTool.get("kafka.sink.topic"),
                new MetricSchema()
                )).name("flink-connectors-kafka")
                .setParallelism(parameterTool.getInt("stream.sink.parallelism"));

        env.execute("flink learning connectors kafka");
    }
}

运行结果

启动程序，查看运行结果，不段执行上面命令，查看是否有新的 topic 出来：插件

执行命令能够查看该 topic 的信息：code

bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic metric-test

分析

上面代码咱们使用 Flink Kafka Producer 只传了三个参数：brokerList、topicId、serializationSchema（序列化）

其实也能够传入多个参数进去，如今有的参数用的是默认参数，由于这个内容比较多，后面能够抽出一篇文章单独来说。

总结

本篇文章写了 Flink 读取其余 Kafka 集群的数据，而后写入到本地的 Kafka 上。我在 Flink 这层没作什么数据转换，只是原样的将数据转发了下，若是大家有什么其余的需求，是能够在 Flink 这层将数据进行各类转换操做，好比这篇文章中的一些转换：《从0到1学习Flink》—— Flink Data transformation(转换)，而后将转换后的数据发到 Kafka 上去。

本文原创地址是: http://www.54tianzhisheng.cn/2019/01/06/Flink-Kafka-sink/ , 未经容许禁止转载。