使用Flink实现索引数据到Elasticsearch
使用Flink实现索引数据到Elasticsearch
使用Flink处理数据时,可以基于Flink提供的批式处理(Batch Processing)和流式处理(Streaming Processing)API来实现,分别能够满足不同场景下应用数据的处理。这两种模式下,输入处理都被抽象为Source Operator,包含对应输入数据的处理逻辑;输出处理都被抽象为Sink Operator,包含了对应输出数据的处理逻辑。这里,我们只关注输出的Sink Operator实现。
Flink批式处理模式,运行Flink Batch Job时作用在有界的输入数据集上,所以Job运行的时间是有时限的,一旦Job运行完成,对应的整个数据处理应用就已经结束,比如,输入是一个数据文件,或者一个Hive SQL查询对应的结果集,等等。在批式处理模式下处理数据的输出时,主要需要实现一个自定义的OutputFormat,然后基于该OutputFormat来构建一个Sink,下面看下OutputFormat接口的定义,如下所示:
1
2
3
4
5
6
7
|
@Public
public interface OutputFormat<IT> extends Serializable {
void configure(Configuration parameters);
void open( int taskNumber, int numTasks) throws IOException;
void writeRecord(IT record) throws IOException;
void close() throws IOException;
}
|
上面,configure()方法用来配置一个OutputFormat的一些输出参数;open()方法用来实现与外部存储系统建立连接;writeRecord()方法用来实现对Flink Batch Job处理后,将数据记录输出到外部存储系统。开发Batch Job时,通过调用DataSet的output()方法,参数值使用一个OutputFormat的具体实现即可。后面,我们会基于Elasticsearch来实现上面接口中的各个方法。
Flink流式处理模式,运行Flink Streaming Job时一般输入的数据集为流数据集,也就是说输入数据元素会持续不断地进入到Streaming Job的处理过程中,但你仍然可以使用一个HDFS数据文件作为Streaming Job的输入,即使这样,一个Flink Streaming Job启动运行后便会永远运行下去,除非有意外故障或有计划地操作使其终止。在流式处理模式下处理数据的输出时,我们需要是实现一个SinkFunction,它指定了如下将流数据处理后的结果,输出到指定的外部存储系统中,下面看下SinkFunction的接口定义,如下所示:
01
02
03
04
05
06
07
08
09
10
11
12
13
14
15
|
@Public
public interface SinkFunction<IN> extends Function, Serializable {
@Deprecated
default void invoke(IN value) throws Exception {}
default void invoke(IN value, Context context) throws Exception {
invoke(value);
}
@Public
interface Context<T> {
long currentProcessingTime();
long currentWatermark();
Long timestamp();
}
}
|
通过上面接口可以看到,需要实现一个invoke()方法,实现该方法来将一个输入的IN value输出到外部存储系统中。一般情况下,对一些主流的外部存储系统,Flink实现了一下内置(社区贡献)的SinkFunction,我们只需要配置一下就可以直接使用。而且,对于Streaming Job来说,实现的SinkFunction比较丰富一些,可以减少自己开发的工作量。开发Streaming Job时,通过调用DataStream的addSink()方法,参数是一个SinkFlink的具体实现。
下面,我们分别基于批式处理模式和批式处理模式,分别使用或实现对应组件将Streaming Job和Batch Job的处理结果输出到Elasticsearch中:
基于Flink DataSteam API实现
在开发基于Flink的应用程序过程中,发现Flink Streaming API对Elasticsearch的支持还是比较好的,比如,如果想要从Kafka消费事件记录,经过处理最终将数据记录索引到Elasticsearch 5.x,可以直接在Maven的POM文件中添加如下依赖即可:
1
2
3
4
5
|
< dependency >
< groupId >org.apache.flink</ groupId >
< artifactId >flink-connector-elasticsearch5_2.11</ artifactId >
< version >1.5.3</ version >
</ dependency >
|
我们使用Flink Streaming API来实现将流式数据处理后,写入到Elasticsearch中。其中,输入数据源是Kafka中的某个Topic;输出处理结果到lasticsearch中,我们使用使用Transport API的方式来连接Elasticsearch,需要指定Transport地址和端口。具体实现,对应的Scala代码,如下所示:
01
02
03
04
05
06
07
08
09
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
|
def main(args : Array[String]) : Unit = {
// parse input arguments
val params = ParameterTool.fromArgs(args)
if (params.getNumberOfParameters < 9 ) {
val cmd = getClass.getName
println( "Missing parameters!\n"
+ "Usage: " + cmd
+ " --input-topic <topic> "
+ "--es-cluster-name <es cluster name> "
+ "--es-transport-addresses <es address> "
+ "--es-port <es port> "
+ "--es-index <es index> "
+ "--es-type <es type> "
+ "--bootstrap.servers <kafka brokers> "
+ "--zookeeper.connect <zk quorum> "
+ "--group.id <some id> [--prefix <prefix>]" )
return
}
val env = StreamExecutionEnvironment.getExecutionEnvironment
val kafkaConsumer = new FlinkKafkaConsumer 010 [String](
params.getRequired( "input-topic" ),
new SimpleStringSchema(),
params.getProperties
)
val dataStream = env
.addSource(kafkaConsumer)
.filter(! _ .isEmpty)
val esClusterName = params.getRequired( "es-cluster-name" )
val esAddresses = params.getRequired( "es-transport-addresses" )
val esPort = params.getInt( "es-port" , 9300 )
val transportAddresses = new java.util.ArrayList[InetSocketAddress]
val config = new java.util.HashMap[String, String]
config.put( "cluster.name" , esClusterName)
// This instructs the sink to emit after every element, otherwise they would be buffered
config.put( "bulk.flush.max.actions" , "100" )
esAddresses.split( "," ).foreach(address = > {
transportAddresses.add( new InetSocketAddress(InetAddress.getByName(address), esPort))
})
val esIndex = params.getRequired( "es-index" )
val esType = params.getRequired( "es-type" )
val sink = new ElasticsearchSink(config, transportAddresses, new ElasticsearchSinkFunction[String] {
def createIndexRequest(element : String) : IndexRequest = {
return Requests.indexRequest()
.index(esIndex)
.` type `(esType)
.source(element)
}
override def process(t : String, runtimeContext : RuntimeContext, requestIndexer : RequestIndexer) : Unit = {
requestIndexer.add(createIndexRequest(t))
}
})
dataStream.addSink(sink)
val jobName = getClass.getSimpleName
env.execute(jobName)
}
|
上面有关数据索引到Elasticsearch的处理中, 最核心的就是创建一个ElasticsearchSink,然后通过DataStream的API调用addSink()添加一个Sink,实际是一个SinkFunction的实现,可以参考Flink对应DataStream类的addSink()方法代码,如下所示:
1
2
|
def addSink(sinkFunction : SinkFunction[T]) : DataStreamSink[T] =
stream.addSink(sinkFunction)
|
基于Flink DataSet API实现
目前,Flink还没有在Batch处理模式下实现对应Elasticsearch对应的Connector,需要自己根据需要实现,所以我们基于Flink已经存在的Streaming处理模式下已经实现的Elasticsearch Connector对应的代码,经过部分修改,可以直接拿来在Batch处理模式下,将数据记录批量索引到Elasticsearch中。
我们基于Flink 1.6.1版本,以及Elasticsearch 6.3.2版本,并且使用Elasticsearch推荐的High Level REST API来实现(为了复用Flink 1.6.1中对应的Streaming处理模式下的Elasticsearch 6 Connector实现代码,我们选择使用该REST Client),需要在Maven的POM文件中添加如下依赖:
01
02
03
04
05
06
07
08
09
10
|
<dependency>
<groupId>org.elasticsearch</groupId>
<artifactId>elasticsearch</artifactId>
<version>6.3.2</version>
</dependency>
<dependency>
<groupId>org.elasticsearch.client</groupId>
<artifactId>elasticsearch-rest-high-level-client</artifactId>
<version>6.3.2</version>
</dependency>
|
我们实现的各个类的类图及其关系,如下图所示:
如果熟悉Flink Streaming处理模式下Elasticsearch对应的Connector实现,可以看到上面的很多类都在org.apache.flink.streaming.connectors.elasticsearch包里面存在,其中包括批量向Elasticsearch中索引数据(内部实现了使用BulkProcessor)。上图中引入的ElasticsearchApiCallBridge,目的是能够实现对Elasticsearch不同版本的支持,只需要根据Elasticsearch不同版本中不同Client实现,进行一些适配,上层抽象保持不变。
如果需要在Batch处理模式下批量索引数据到Elasticsearch,可以直接使用ElasticsearchOutputFormat即可实现。但是创建ElasticsearchOutputFormat,需要几个参数:
1
2
3
4
5
6
7
8
|
private ElasticsearchOutputFormat(
Map<String, String> bulkRequestsConfig,
List<HttpHost> httpHosts,
ElasticsearchSinkFunction<T> elasticsearchSinkFunction,
DocWriteRequestFailureHandler failureHandler,
RestClientFactory restClientFactory) {
super ( new Elasticsearch6ApiCallBridge(httpHosts, restClientFactory), bulkRequestsConfig, elasticsearchSinkFunction, failureHandler);
}
|
当然,我们可以通过代码中提供的Builder来非常方便的创建一个ElasticsearchOutputFormat。下面,我们看下我们Flink Batch Job实现逻辑。
- 实现ElasticsearchSinkFunction
我们需要实现ElasticsearchSinkFunction接口,实现一个能够索引数据到Elasticsearch中的功能,代码如下所示:
01
02
03
04
05
06
07
08
09
10
11
12
13
14
15
16
|
final ElasticsearchSinkFunction<String> elasticsearchSinkFunction = new ElasticsearchSinkFunction<String>() {
@Override
public void process(String element, RuntimeContext ctx, RequestIndexer indexer) {
indexer.add(createIndexRequest(element, parameterTool));
}
private IndexRequest createIndexRequest(String element, ParameterTool parameterTool) {
LOG.info( "Create index req: " + element);
JSONObject o = JSONObject.parseObject(element);
return Requests.indexRequest()
.index(parameterTool.getRequired( "es-index" ))
.type(parameterTool.getRequired( "es-type" ))
.source(o);
}
};
|
上面代码,主要是把一个将要输出的数据记录,通过RequestIndexer来实现索引到Elasticsearch中。
- 读取Elasticsearch配置参数
配置连接Elasticsearch的参数。从程序输入的ParameterTool中读取Elasticsearch相关的配置:
01
02
03
04
05
06
07
08
09
10
11
12
13
|
String esHttpHosts = parameterTool.getRequired( "es-http-hosts" );
LOG.info( "Config: esHttpHosts=" + esHttpHosts);
int esHttpPort = parameterTool.getInt( "es-http-port" , 9200 );
LOG.info( "Config: esHttpPort=" + esHttpPort);
final List<HttpHost> httpHosts = Arrays.asList(esHttpHosts.split( "," ))
.stream()
.map(host -> new HttpHost(host, esHttpPort, "http" ))
.collect(Collectors.toList());
int bulkFlushMaxSizeMb = parameterTool.getInt( "bulk-flush-max-size-mb" , 10 );
int bulkFlushIntervalMillis = parameterTool.getInt( "bulk-flush-interval-millis" , 10 * 1000 );
int bulkFlushMaxActions = parameterTool.getInt( "bulk-flush-max-actions" , 1 );
|
- 创建ElasticsearchOutputFormat
创建一个我们实现的ElasticsearchOutputFormat,代码片段如下所示:
1
2
3
4
5
6
7
8
|
final ElasticsearchOutputFormat outputFormat = new Builder<>(httpHosts, elasticsearchSinkFunction)
.setBulkFlushBackoff( true )
.setBulkFlushBackoffRetries( 2 )
.setBulkFlushBackoffType(ElasticsearchApiCallBridge.FlushBackoffType.EXPONENTIAL)
.setBulkFlushMaxSizeMb(bulkFlushMaxSizeMb)
.setBulkFlushInterval(bulkFlushIntervalMillis)
.setBulkFlushMaxActions(bulkFlushMaxActions)
.build();
|
上面很多配置项指定了向Elasticsearch中进行批量写入的行为,在ElasticsearchOutputFormat内部会进行设置并创建Elasticsearch6BulkProcessorIndexer,优化索引数据处理的性能。
- 实现Batch Job主控制流程
最后我们就可以构建我们的Flink Batch应用程序了,代码如下所示:
1
2
3
4
5
6
7
8
|
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
env.readTextFile(file)
.filter(line -> !line.isEmpty())
.map(line -> line)
.output(outputFormat);
final String jobName = ImportHDFSDataToES. class .getSimpleName();
env.execute(jobName);
|
我们输入的HDFS文件中,是一些已经加工好的JSON格式记录行,这里为了简单,直接将原始JSON字符串索引到Elasticsearch中,而没有进行更多其他的处理操作。
有关Flink批式处理模式下,Elasticsearch对应的OutputFormat实现的完整代码,可以参考这里:
https://github.com/shirdrn/flink-app-jobs/tree/master/src/main/java/org/shirdrn/flink/connector/batch/elasticsearch。
参考链接
- https://ci.apache.org/projects/flink/flink-docs-release-1.5/dev/connectors/elasticsearch.html
- https://ci.apache.org/projects/flink/flink-docs-release-1.6/dev/batch/#data-sinks
本文基于署名-非商业性使用-相同方式共享 4.0许可协议发布,欢迎转载、使用、重新发布,但务必保留文章署名时延军(包含链接:http://shiyanjun.cn),不得用于商业目的,基于本文修改后的作品务必以相同的许可发布。如有任何疑问,请与我联系。
转载于:https://www.cnblogs.com/bigben0123/p/10065043.html
使用Flink实现索引数据到Elasticsearch相关推荐
- 【Flink】使用Flink实现索引数据到Elasticsearch
1.概述 转载:使用Flink实现索引数据到Elasticsearch 建议看原文 使用Flink处理数据时,可以基于Flink提供的批式处理(Batch Processing)和流式处理(Strea ...
- kinana 清空索引数据_(Elasticsearch)实战Elasticseartch、Logstash、Kibana
1.Elasticsearch的堆栈内存设置建议 image.png 2.elasticsearch.yml设置中文分词器: /usr/share/elasticsearch/bin/elastics ...
- flink 读取文件数据写入ElasticSearch
前言 es是大数据存储的必备中间件之一,通过flink可以读取来自日志文件,kafka等外部数据源的数据,然后写入到es中,本篇将通过实例演示下完整的操作过程: 一.前置准备 1.提前搭建并开启es服 ...
- 《从0到1学习Flink》—— Flink 写入数据到 ElasticSearch
前言 前面 FLink 的文章中我们已经介绍了说 Flink 已经有很多自带的 Connector. 1.<从0到1学习Flink>-- Data Source 介绍 2.<从0到1 ...
- python将ElasticSearch索引数据读入pandas dataframe实战
python将ElasticSearch索引数据读入pandas dataframe实战 # 导入基础包和库 import pandas as pdpd.set_option('display.max ...
- 从关系数据库到Elasticsearch的索引数据– 1
Elasticsearch提供强大的搜索功能,并支持数据的分片和复制. 因此,我们希望将数据库中可用的数据索引到Elasticsearch中. 有多种方法可以将数据索引到Elasticsearch中: ...
- 【ElasticSearch】在 ELASTICSEARCH 中使用管道重新索引数据 pipeline
1.概述 翻译:https://cinhtau.net/2017/05/01/reindex-data-with-pipeline/ 数据并不总是干净的.根据它的生成方式,数字可能会在 JSON 正文 ...
- elasticsearch备份与恢复4_使用ES-Hadoop将ES中的索引数据写入HDFS中
背景知识见链接:elasticsearch备份与恢复3_使用ES-Hadoop将HDFS数据写入Elasticsearch中 项目参考<Elasticsearch集成Hadoop最佳实践> ...
- 详述 Elasticsearch 通过范围条件查询索引数据的方法
文章目录 情景 查询方法 通过命令实现范围查询 通过 API 实现范围查询 情景 在使用 Elasticsearch 的时候,我们可能会遇到需要以范围为条件查询索引数据的需求.有两种方法可以实现我们的 ...
- Elasticsearch 索引数据多了怎么办,如何调优,部署?
1 动态索引层面 基于模板+时间+rollover api滚动创建索引,举例:设计阶段定义:blog索引的模板格式为:blog_index_时间戳的形式,每天递增数据.这样做的好处:不至于数据量激增导 ...
最新文章
- Nature指数2021亚太区排名:7所中国高校挺进前10!看看有没有你的母校?
- jquery 1.9里面已经删除了toggle(fn1, fn2)函数:
- swoole安装全纪录
- sqlserver 中的exec问题
- 对称二叉树—leetcode101
- SRM 440(1-250pt, 1-500pt)
- HDU - 6267 (概论/找规律/递推)
- android密码可见不可见的光标控制,Android EditText 在设置为输入密码的时候 密码是否可见 光标在最后显示...
- linux文件系统 环形结构图,环形缓冲器(转)
- ubuntu19.04安装pip3以及virtualenv和virtualenvwrapper
- 8个习惯让你减肥不用节食 - 生活至上,美容至尚!
- 创建线程的3种方式,线程池的参数详解
- 联众打码平台接口调用(初版)
- 用java判定三角形_人教版初中数学八年级上册“角角边”判定三角形全等公开课优质课课件教案视频...
- c语言题库系统的实现,C语言试题库组卷系统的研究与实现
- Excel如何合并两个单元格内容
- Android-使用HttpURLConnection实现多线程下载
- 第六期 Wiggler调试路由器 《路由器就是开发板》
- 【Linux Centos6/7连接Oracle11g数据库,提示:ORA-12514 TNS 监听程序当前无法识别连接描述符中请求服务】
- 【DS实践 | Coursera】Assignment 3 | Applied Plotting, Charting Data Representation in Python
热门文章
- getParameter和getAttribute区别(超详细分析)
- 2021-06-22列表样式与背景图片
- 2021-06-13list map set 并发问题
- 2021年各省高考试成绩查询,2021年各省高考成绩查询时间 什么时候出分
- 八卦图代码matlab,12行javascript代码绘制一个八卦图_javascript技巧
- 添加key_所写为方便日后查阅(添加SSHkey)
- 华为 台积电 高通申请_华为表态愿意合作,台积电送来“神助攻”,高通:我太难了...
- 利用后中遍历结果,重构二叉树
- nginx 站点使用try_files配置案例
- Java 面试 ——可变参数、初始化数据块、设计秒杀系统