Elasticsearch 是一个分布式的、RESTful 风格的搜索和数据分析引擎。Apache Flink 官方提供了 Elasticsearch 的连接器，用于向 ElasticSearch 中写入数据,可提供 至少一次 的处理语义。

ElasticsearchSink 使用 TransportClient（6.x 之前）或者 RestHighLevelClient（6.x 开始）和 Elasticsearch 集群进行通信，Apache StreamPark 对 flink-connector-elasticsearch6 进一步封装，屏蔽开发细节，简化 Elasticsearch6 及以上的写入操作。

因为 Flink Connector Elasticsearch 不同版本之间存在冲突，StreamPark 暂时仅支持 Elasticsearch6 及以上的写入操作，如需写入 Elasticsearch5 集群，需要使用者排除 flink-connector-elasticsearch6 依赖，引入 flink-connector-elasticsearch5 依赖。创建 org.apache.flink.streaming.connectors.elasticsearch5.ElasticsearchSink 实例写入数据。

Elasticsearch 写入依赖

Elasticsearch 版本不同依赖 Flink Connector Elasticsearch 不同,以下信息来源flink-docs-release-1.14文档: Elasticsearch 5.x Maven依赖

  1. <dependency>
  2.     <groupId>org.apache.flink</groupId>
  3.     <artifactId>flink-connector-elasticsearch5_2.11</artifactId>
  4.     <version>1.14.3</version>
  5. </dependency>

Elasticsearch 6.x Maven依赖

  1. <dependency>
  2.     <groupId>org.apache.flink</groupId>
  3.     <artifactId>flink-connector-elasticsearch6_2.11</artifactId>
  4.     <version>1.14.3</version>
  5. </dependency>

Elasticsearch 7.x及以上 Maven依赖

  1. <dependency>
  2.     <groupId>org.apache.flink</groupId>
  3.     <artifactId>flink-connector-elasticsearch7_2.11</artifactId>
  4.     <version>1.14.3</version>
  5. </dependency>

基于官网的Elasticsearch写入数据

以下代码摘自官方文档

java, Elasticsearch 6.x 及以上

  1. import org.apache.flink.api.common.functions.RuntimeContext;
  2. import org.apache.flink.streaming.api.datastream.DataStream;
  3. import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkFunction;
  4. import org.apache.flink.streaming.connectors.elasticsearch.RequestIndexer;
  5. import org.apache.flink.streaming.connectors.elasticsearch6.ElasticsearchSink;
  7. import org.apache.http.HttpHost;
  8. import org.elasticsearch.action.index.IndexRequest;
  9. import org.elasticsearch.client.Requests;
  11. import java.util.ArrayList;
  12. import java.util.HashMap;
  13. import java.util.List;
  14. import java.util.Map;
  16. DataStream<String> input = ...;
  18. List<HttpHost> httpHosts = new ArrayList<>();
  19. httpHosts.add(new HttpHost("127.0.0.1", 9200, "http"));
  20. httpHosts.add(new HttpHost("10.2.3.1", 9200, "http"));
  22. // 使用 ElasticsearchSink.Builder 创建 ElasticsearchSink
  23. ElasticsearchSink.Builder<String> esSinkBuilder = new ElasticsearchSink.Builder<>(
  24.     httpHosts,
  25.     new ElasticsearchSinkFunction<String>() {
  26.         public IndexRequest createIndexRequest(String element) {
  27.             Map<String, String> json = new HashMap<>();
  28.             json.put("data", element);
  30.             return Requests.indexRequest()
  31.                     .index("my-index")
  32.                     .type("my-type")
  33.                     .source(json);
  34.         }
  36.         @Override
  37.         public void process(String element, RuntimeContext ctx, RequestIndexer indexer) {
  38.             indexer.add(createIndexRequest(element));
  39.         }
  40.     }
  41. );
  43. // 批量请求的配置；下面的设置使 sink 在接收每个元素之后立即提交，否则这些元素将被缓存起来
  44. esSinkBuilder.setBulkFlushMaxActions(1);
  46. // 为内部创建的 REST 客户端提供一个自定义配置信息的 RestClientFactory
  47. esSinkBuilder.setRestClientFactory(
  48.   restClientBuilder -> {
  49.     restClientBuilder.setDefaultHeaders(...)
  50.     restClientBuilder.setMaxRetryTimeoutMillis(...)
  51.     restClientBuilder.setPathPrefix(...)
  52.     restClientBuilder.setHttpClientConfigCallback(...)
  53.   }
  54. );
  56. // 最后，构建并添加 sink 到作业管道中
  57. input.addSink(esSinkBuilder.build());

scala, Elasticsearch 6.x 及以上

  2. import org.apache.flink.api.common.functions.RuntimeContext
  3. import org.apache.flink.streaming.api.datastream.DataStream
  4. import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkFunction
  5. import org.apache.flink.streaming.connectors.elasticsearch.RequestIndexer
  6. import org.apache.flink.streaming.connectors.elasticsearch6.ElasticsearchSink
  8. import org.apache.http.HttpHost
  9. import org.elasticsearch.action.index.IndexRequest
  10. import org.elasticsearch.client.Requests
  12. import java.util.ArrayList
  13. import java.util.List
  15. val input: DataStream[String] = ...
  17. val httpHosts = new java.util.ArrayList[HttpHost]
  18. httpHosts.add(new HttpHost("127.0.0.1", 9200, "http"))
  19. httpHosts.add(new HttpHost("10.2.3.1", 9200, "http"))
  21. val esSinkBuilder = new ElasticsearchSink.Builder[String](
  22.   httpHosts,
  23.   new ElasticsearchSinkFunction[String] {
  24.      def process(element: String, ctx: RuntimeContext, indexer: RequestIndexer) {
  25.           val json = new java.util.HashMap[String, String]
  26.           json.put("data", element)
  28.           val rqst: IndexRequest = Requests.indexRequest
  29.             .index("my-index")
  30.             .`type`("my-type")
  31.             .source(json)
  33.           indexer.add(rqst)
  34.      }
  35.   }
  36. )
  38. // 批量请求的配置；下面的设置使 sink 在接收每个元素之后立即提交，否则这些元素将被缓存起来
  39. esSinkBuilder.setBulkFlushMaxActions(1)
  41. // 为内部创建的 REST 客户端提供一个自定义配置信息的 RestClientFactory
  42. esSinkBuilder.setRestClientFactory(new RestClientFactory {
  43.   override def configureRestClientBuilder(restClientBuilder: RestClientBuilder): Unit = {
  44.        restClientBuilder.setDefaultHeaders(...)
  45.        restClientBuilder.setMaxRetryTimeoutMillis(...)
  46.        restClientBuilder.setPathPrefix(...)
  47.        restClientBuilder.setHttpClientConfigCallback(...)
  48.   }
  49. })
  51. // 最后，构建并添加 sink 到作业管道中
  52. input.addSink(esSinkBuilder.build)

以上创建ElasticsearchSink添加参数非常的不灵敏。StreamPark使用约定大于配置、自动配置的方式只需要配置es 连接参数、flink运行参数，StreamPark 会自动组装source和sink，极大的简化开发逻辑，提升开发效率和维护性。

Apache StreamPark™ 写入 Elasticsearch

ESSink 在启用 Flink checkpoint 后，保证至少一次将操作请求发送到 Elasticsearch 集群。

1. 配置策略和连接信息

  1. #redis sink 配置
  2. #  必填参数，多个节点使用 host1:port, host2:port,
  3. host: localhost:9200
  4. #  选填参数
  5. #  es:
  6. #    disableFlushOnCheckpoint: true #默认为false
  7. #    auth:
  8. #    user:
  9. #      password:
  10. #    rest:
  11. #      max.retry.timeout:
  12. #      path.prefix:
  13. #      content.type:
  14. #    connect:
  15. #      request.timeout:
  16. #      timeout:
  17. #    cluster.name: elasticsearch
  18. #  client.transport.sniff:
  19. #  bulk.flush.:

2. 写入Elasticsearch

用 StreamPark 写入Elasticsearch非常简单,代码如下:

Scala

  1. import org.apache.streampark.flink.core.scala.FlinkStreaming
  2. import org.apache.streampark.flink.core.scala.sink.ESSink
  3. import org.apache.streampark.flink.core.scala.util.ElasticSearchUtils
  4. import org.apache.flink.api.scala._
  5. import org.elasticsearch.action.index.IndexRequest
  6. import org.json4s.DefaultFormats
  7. import org.json4s.jackson.Serialization
  9. import java.util.Date
  11. object ConnectorApp extends FlinkStreaming {
  14.   implicit lazy val formats: DefaultFormats.type = org.json4s.DefaultFormats
  16.   override def handle(): Unit = {
  17.     val ds = context.fromCollection(List(
  18.       OrderEntity(1, 1, 11.3d, 3.1d, new Date()),
  19.       OrderEntity(2, 1, 12.3d, 3.2d, new Date()),
  20.       OrderEntity(3, 1, 13.3d, 3.3d, new Date()),
  21.       OrderEntity(4, 1, 14.3d, 3.4d, new Date()),
  22.       OrderEntity(5, 1, 15.3d, 7.5d, new Date()),
  23.       OrderEntity(6, 1, 16.3d, 3.6d, new Date()),
  24.       OrderEntity(7, 1, 17.3d, 3.7d, new Date())
  25.     ))
  27.     // es sink.........
  29.     //1)定义 Index的写入规则
  30.     implicit def indexReq(x: OrderEntity): IndexRequest = ElasticSearchUtils.indexRequest(
  31.       "flink_order",
  32.       "_doc",
  33.       s"${x.id}_${x.time.getTime}",
  34.       Serialization.write(x)
  35.     )
  36.     //3)定义esSink并下沉=数据. done
  37.     ESSink().sink6[OrderEntity](ds)
  38.   }
  41.   case class OrderEntity(id: Int, num: Int, price: Double, gmv: Double, time: Date) extends Serializable
  43. }

Flink ElasticsearchSinkFunction可以执行多种类型请求，如（DeleteRequest、 UpdateRequest、IndexRequest）,StreamPark也对以上功能进行了支持，对应方法如下：

  1. import org.apache.streampark.flink.core.scala.StreamingContext
  2. import org.apache.flink.streaming.api.datastream.DataStreamSink
  3. import org.apache.flink.streaming.api.scala.DataStream
  4. import org.apache.flink.streaming.connectors.elasticsearch.ActionRequestFailureHandler
  5. import org.apache.flink.streaming.connectors.elasticsearch.util.RetryRejectedExecutionFailureHandler
  6. import org.apache.flink.streaming.connectors.elasticsearch6.RestClientFactory
  7. import org.elasticsearch.action.delete.DeleteRequest
  8. import org.elasticsearch.action.index.IndexRequest
  9. import org.elasticsearch.action.update.UpdateRequest
  11. import java.util.Properties
  12. import scala.annotation.meta.param
  15. class ESSink(@(transient@param) context: StreamingContext,
  16.              property: Properties = new Properties(),
  17.              parallelism: Int = 0,
  18.              name: String = null,
  19.              uid: String = null) {
  21.   /**
  22.    * for ElasticSearch6
  23.    *
  24.    * @param stream
  25.    * @param suffix
  26.    * @param restClientFactory
  27.    * @param failureHandler
  28.    * @param f
  29.    * @tparam T
  30.    * @return
  31.    */
  32.   def sink6[T](stream: DataStream[T],
  33.                suffix: String = "",
  34.                restClientFactory: RestClientFactory = null,
  35.                failureHandler: ActionRequestFailureHandler = new RetryRejectedExecutionFailureHandler)
  36.               (implicit f: T => IndexRequest): DataStreamSink[T] = {
  38.     new ES6Sink(context, property, parallelism, name, uid).sink[T](stream, suffix, restClientFactory, failureHandler, f)
  39.   }
  43.   def update6[T](stream: DataStream[T],
  44.                suffix: String = "",
  45.                restClientFactory: RestClientFactory = null,
  46.                failureHandler: ActionRequestFailureHandler = new RetryRejectedExecutionFailureHandler)
  47.               ( f: T => UpdateRequest): DataStreamSink[T] = {
  49.     new ES6Sink(context, property, parallelism, name, uid).sink[T](stream, suffix, restClientFactory, failureHandler, f)
  50.   }
  53.   def delete6[T](stream: DataStream[T],
  54.                suffix: String = "",
  55.                restClientFactory: RestClientFactory = null,
  56.                failureHandler: ActionRequestFailureHandler = new RetryRejectedExecutionFailureHandler)
  57.               ( f: T => DeleteRequest): DataStreamSink[T] = {
  59.     new ES6Sink(context, property, parallelism, name, uid).sink[T](stream, suffix, restClientFactory, failureHandler, f)
  60.   }
  62. }

启用 Flink checkpoint 后，Flink Elasticsearch Sink 保证至少一次将操作请求发送到 Elasticsearch 集群。 这是通过在进行 checkpoint 时等待 BulkProcessor 中所有挂起的操作请求来实现。 这有效地保证了在触发 checkpoint 之前所有的请求被 Elasticsearch 成功确认，然后继续处理发送到 sink 的记录。 用户想要禁用刷新，可以配置disableFlushOnCheckpoint为true来实现，实质上意味着 sink 将不再提供任何可靠的交付保证，即使启用了作业拓扑的 checkpoint。

其他配置

处理失败的 Elasticsearch 请求

Elasticsearch 操作请求可能由于多种原因而失败，可以通过实现ActionRequestFailureHandler来指定失败处理逻辑，见 官方文档处理失败的 Elasticsearch 请求 单元

配置内部批量处理器

es内部BulkProcessor可以进一步配置其如何刷新缓存操作请求的行为详细查看官方文档配置内部批量处理器 单元

Apache StreamPark™配置

其他的所有的配置都必须遵守 StreamPark 配置,具体可配置项和各个参数的作用请参考项目配置