Elasticsearch:Data pipeline: Kafka => Flink => Elasticsearch

2021年10月13日   |   by mebius

昨天我写了一篇关于 Flink 和 Elasticsearch 方面的文章 “Elasticsearch:使用 Apache Flink、Elasticsearch 打造实时事件处理及搜索”。在那篇文章中,我详细地描述了如何把 Flink 的数据写入到 Elasticsearch 中。在实际的使用过程中,我们可能经常会使用到 Kafka,并把 Kafka 里的数据经过 Flink 处理后再写入到 Elasticsearch 中。在今天的文章中,我将来详述如何实现。

%title插图%num

安装

如果你还没有安装好自己的 Elastic Stack 及 Flink 的话,请参阅我之前的文章 “Elasticsearch:使用 Apache Flink、Elasticsearch 打造实时事件处理及搜索”。按照里面的步骤,我们分别进行安装,并启动 Elasticsearch,Kibana 及 Flink。

接下来,我们需要安装 Kafka。

Linux

我们可以参考文章 “Elastic:使用 Kafka 部署 Elastic Stack” 来安装 Kafka。 简单地说就是:

Kafka 使用 ZooKeeper 来维护配置信息和同步,因此在设置 Kafka 之前,我们需要先安装 ZooKeeper:

sudo apt-get install zookeeperd

接下来,让我们下载并解压缩 Kafka:

wget https://apache.mivzakim.net/kafka/2.4.0/kafka_2.13-2.4.0.tgz
tar -xzvf kafka_2.13-2.4.0.tgz
sudo cp -r kafka_2.13-2.4.0 /opt/kafka

现在,我们准备运行 Kafka,我们将使用以下脚本进行操作:

sudo /opt/kafka/bin/kafka-server-start.sh /opt/kafka/config/server.properties

你应该开始在控制台中看到一些 INFO 消息:

%title插图%num

我们接下来打开另外一个控制台中,并为 Apache 日志创建一个主题:

/opt/kafka/bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic mytopic

我们创建了一个叫做 mytopic 的 topic。上面的命令将返回如下的结果:

Created topic apache.

我们现在已经完全为开始管道做好了准备。

macOS

MacOS 的安装也非常的直接。我们直接使用如下的命令来进行分别安装 zookeeper 及 Kafka:

brew installl zookeeper
brew install kafka

它们的安装目录分别位于如下的位置:

/usr/local/bin/zkServer
/usr/local/opt/kafka/bin/kafka-server-start

我们使用如下的命令来启动 zookeeper:

$ zkServer start
ZooKeeper JMX enabled by default
Using config: /usr/local/etc/zookeeper/zoo.cfg
Starting zookeeper ... STARTED

我们使用如下的命令来启动 Kafka:

$ /usr/local/opt/kafka/bin/kafka-server-start /usr/local/etc/kafka/server.properties

当然你也可以启动 Kafka 为一个服务,而不是想上面的那样值运行一次:

brew services start kafka

接下来,我们使用如下的命令来创建一个叫做 mytopic 的 topic:

kafka-topics --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic mytopic
$ kafka-topics --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic mytopic
Created topic mytopic.

这样我们就准备好了。

创建演示例子

接下来,我们将使用 Java 来构建一个展示的例子。它使用 API 来访问 Flink。如上所示,我们将使用 Flink 的 enviornment,source,transform 及 sink APIs 来构建我们的应用。为了方便大家学习,我已经把我的项目上传到 github 了。你需要使用如下的命令来进行下载:

git clone https://github.com/liu-xiao-guo/kafka-flink-elasticsearch

你可以使用你自己喜欢的 IDE 来创建一个新的项目来开始。

KafkaElasticsearch.java

import java.util.*;
import java.util.logging.Logger;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.TimeCharacteristic;
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.elasticsearch.ElasticsearchSinkFunction;
import org.apache.flink.streaming.connectors.elasticsearch.RequestIndexer;
import org.apache.flink.streaming.connectors.elasticsearch7.ElasticsearchSink;
import org.apache.flink.streaming.util.serialization.SimpleStringSchema;
import org.apache.http.HttpHost;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.Requests;

public class KafkaElasticsearch {
    // private static final Logger log = Logger.getLogger(KafkaElasticsearch.class.getSimpleName());
    private static final Logger log = Logger.getLogger(KafkaElasticsearch.class.getSimpleName());

    public static void main(String[] args) throws Exception {
        System.out.println("Starting ....");

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // Configure kafka parameters
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        env.enableCheckpointing(5000);
        KafkaSource source = KafkaSource.builder()
                .setBootstrapServers("localhost:9092")
                .setTopics("mytopic")
                .setGroupId("my-group")
                .setStartingOffsets(OffsetsInitializer.earliest())
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .build();

        DataStreamSource kafkaData  = env.fromSource(source,
                WatermarkStrategy.noWatermarks(), "Kafka Source");
        DattgcodeaStream transSource = kafktgcodeaData.map(new MapFunction() {
            @Override
            public String map(String s) {
                System.out.println("Received message: " + s);
                return s;
            }
        });

        try {
            List hosts = new ArrayList();
            hosts.add(new HttpHost("localhost", 9200, "http"));
            ElasticsearchSink.Builder builder = new ElasticsearchSink.Builder(htgcodeosts,
                    new ElasticsearchSinkFunction() {
                        @Override
                        public void process(String s, RuntimeContext runtimeContext, RequestIndexer requestIndexer) {
                            Map jsonMap = new HashMap();
                            jsonMap.put("message", s);
                            IndexRequest indexRequest = Requests.indexRequest();
                            indexRequest.index("kafka-flink");
                            // indexRequest.id("1000");
                            indexRequest.source(jsonMap);
                            requestIndexer.add(indexRequest);
                        }
                    });

            // Define a sink
            builder.setBulkFlushMaxActions(1);
            transSource.addSink(builder.build());

            // Execute the transform
            env.execute("flink-es");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

整个代码的设计非常简单。如果你有什么不清楚的,请参阅我之前的文章“Elasticsearch:使用 Apache Flink、Elasticsearch 打造实时事件处理及搜索” 以及 Apache Kafka Connector

我们运行上面的应用,并在一个 terminal 中使用 Kafka 安装自带的工具向 Kafka 的 topic “mytopic” 进行写入数据:

kafka-console-producer --broker-list localhost:9092 --topic mytopic

%title插图%num

我们在 Kibana 中查看索引kafka-flink:

GET kafka-flink/_search

%title插图%num

从上面,我们可以看出来我们发送到 mytopic 的数据最终到达 Elasticsearch。

参考:

【1】Kafka | Apache Flink

文章来源于互联网:Elasticsearch:Data pipeline: Kafka => Flink => Elasticsearch

相关推荐: Elasticsearch:配置 TLS/SSL 和 PKI 身份验证

当为使用生产许可证运行的集群启用 Elasticsearch 安全性时,必须使用 TLS/SSL 进行传输通信,并且必须正确设置。此外,一旦启用安全性,与 Elasticsearch 集群的所有通信都必须经过身份验证,包括来自 Kibana 和/或应用程序服务…