使用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接口的定义,如下所示:
@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的接口定义,如下所示:
@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文件中添加如下依赖即可:
<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代码,如下所示:
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 FlinkKafkaConsumer010[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()方法代码,如下所示:
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文件中添加如下依赖:
<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,需要几个参数:
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中的功能,代码如下所示:
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相关的配置:
创建ElasticsearchOutputFormat
创建一个我们实现的ElasticsearchOutputFormat,代码片段如下所示:
上面很多配置项指定了向Elasticsearch中进行批量写入的行为,在ElasticsearchOutputFormat内部会进行设置并创建Elasticsearch6BulkProcessorIndexer,优化索引数据处理的性能。
实现Batch Job主控制流程
最后我们就可以构建我们的Flink Batch应用程序了,代码如下所示:
我们输入的HDFS文件中,是一些已经加工好的JSON格式记录行,这里为了简单,直接将原始JSON字符串索引到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
低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。
持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。
转载内容版权归作者及来源网站所有,本站原创内容转载请注明来源。
- 上一篇
Apache Flink:Keyed Window与Non-Keyed Window
5万人关注的大数据成神之路,不来了解一下吗?5万人关注的大数据成神之路,真的不来了解一下吗?5万人关注的大数据成神之路,确定真的不来了解一下吗? Apache Flink中,Window操作在流式数据处理中是非常核心的一种抽象,它把一个无限流数据集分割成一个个有界的Window(或称为Bucket),然后就可以非常方便地定义作用于Window之上的各种计算操作。本文我们主要基于Apache Flink 1.4.0版本,说明Keyed Window与Non-Keyed Window的基本概念,然后分别对与其相关的WindowFunction与WindowAllFunction的类设计进行分析,最后通过编程实践来应用。 基本概念 Flink将Window分为两类,一类叫做Keyed Window,另一类叫做Non-Keyed Window。为了说明这两类Window的不同,我们看下Flink官网给出的,基于这两种类型的Window编写代码的结构说明。基于Keyed Window进行编程,用户代码基本结构如下所示: 基于Non-Keyed Window进行编程,用户代码基本结构如下所示: 上...
- 下一篇
Spark中几种ShuffleWriter的区别你都知道吗?
一.前言 在Spark中有三种shuffle写,分别是BypassMergeSortShuffleWriter、UnsafeShuffleWriter、SortShuffleWriter。分别对应三种不同的shuffleHandle。 这三者和ShuffleHandle的对应关系如下: UnsafeShuffleWriter:SerializedShuffleHandle BypassMergeSortShuffleWriter:BypassMergeSortShuffleHandle, SortShuffleWriter:BaseShuffleHandle 那么这些shuffle写内部的实现细节有何不同,在什么场景下使用什么样的shuffleWriter呢,接下来我们对着三种ShuffleWriter的实现细节做一个比较。 二.不同shuffleHandle的使用时机 不同shuffleWrite的使用其实是根据shuffleHandle来决定的,在构建shuffleDependence时都会构建shuffleHandle,在registerShuffle方法中,有着对shuffle...
相关文章
文章评论
共有0条评论来说两句吧...
文章二维码
点击排行
-
Docker使用Oracle官方镜像安装(12C,18C,19C)
- Springboot2将连接池hikari替换为druid,体验最强大的数据库连接池
- CentOS8编译安装MySQL8.0.19
- Docker快速安装Oracle11G,搭建oracle11g学习环境
- SpringBoot2配置默认Tomcat设置,开启更多高级功能
- MySQL8.0.19开启GTID主从同步CentOS8
- CentOS7,8上快速安装Gitea,搭建Git服务器
- Jdk安装(Linux,MacOS,Windows),包含三大操作系统的最全安装
- SpringBoot2编写第一个Controller,响应你的http请求并返回结果
推荐阅读
最新文章
- Springboot2将连接池hikari替换为druid,体验最强大的数据库连接池
- Docker使用Oracle官方镜像安装(12C,18C,19C)
- CentOS6,CentOS7官方镜像安装Oracle11G
- SpringBoot2整合Redis,开启缓存,提高访问速度
- Jdk安装(Linux,MacOS,Windows),包含三大操作系统的最全安装
- SpringBoot2配置默认Tomcat设置,开启更多高级功能
- SpringBoot2整合MyBatis,连接MySql数据库做增删改查操作
- Hadoop3单机部署,实现最简伪集群
- MySQL8.0.19开启GTID主从同步CentOS8
- SpringBoot2编写第一个Controller,响应你的http请求并返回结果