Elasticsearch:Data pipeline: Kafka => Flink => Elasticsearch
2021年10月13日 | by mebius
昨天我写了一篇关于 Flink 和 Elasticsearch 方面的文章 “Elasticsearch:使用 Apache Flink、Elasticsearch 打造实时事件处理及搜索”。在那篇文章中,我详细地描述了如何把 Flink 的数据写入到 Elasticsearch 中。在实际的使用过程中,我们可能经常会使用到 Kafka,并把 Kafka 里的数据经过 Flink 处理后再写入到 Elasticsearch 中。在今天的文章中,我将来详述如何实现。
安装
如果你还没有安装好自己的 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 消息:
我们接下来打开另外一个控制台中,并为 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
我们在 Kibana 中查看索引kafka-flink:
GET kafka-flink/_search
从上面,我们可以看出来我们发送到 mytopic 的数据最终到达 Elasticsearch。
参考:
文章来源于互联网:Elasticsearch:Data pipeline: Kafka => Flink => Elasticsearch
相关推荐: Elasticsearch:配置 TLS/SSL 和 PKI 身份验证
当为使用生产许可证运行的集群启用 Elasticsearch 安全性时,必须使用 TLS/SSL 进行传输通信,并且必须正确设置。此外,一旦启用安全性,与 Elasticsearch 集群的所有通信都必须经过身份验证,包括来自 Kibana 和/或应用程序服务…