devbox
IT小白
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

使用Flink实现索引数据到Elasticsearch

作者：IT小白 | 2024-05-22 04:15:57

赞

踩

flink setbulkfl

使用Flink实现索引数据到Elasticsearch

 2018-07-28 23:16:36    Yanjun

使用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-Batch-Connector-Elasticsearch
如果熟悉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。

参考链接

Creative Commons License

本文基于署名-非商业性使用-相同方式共享 4.0许可协议发布，欢迎转载、使用、重新发布，但务必保留文章署名时延军（包含链接：http://shiyanjun.cn），不得用于商业目的，基于本文修改后的作品务必以相同的许可发布。如有任何疑问，请与我联系。

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/IT小白/article/detail/606185
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号