Elasticsearch is a distributed, RESTful style search and data analysis engine. Apache Flink officially provides a connector for Elasticsearch, which is used to write data to Elasticsearch, which can provide at least once Semantics.

ElasticsearchSink uses TransportClient (before 6.x) or RestHighLevelClient (starting with 6.x) to communicate with the Elasticsearch cluster. Apache StreamPark further encapsulates Flink-connector-elasticsearch6, shields development details, and simplifies write operations for Elasticsearch6 and above.

Because there are conflicts between different versions of Flink Connector Elasticsearch, StreamPark temporarily only supports write operations of Elasticsearch6 and above. If you wants to using Elasticsearch5, you need to exclude the flink-connector-elasticsearch6 dependency and introduce the flink-connector-elasticsearch5 dependency to create org.apache.flink.streaming.connectors.elasticsearch5.ElasticsearchSink instance writes data.

Dependency of elastic writing

Different Elasticsearch versions rely on the Flink Connector Elasticsearch is not universal, the following information comes from the flink-docs-release-1.14 document:

Elasticsearch 5.x Maven dependencies

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

Elasticsearch 6.x Maven dependencies

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

Elasticsearch 7.x ans above Maven dependencies

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

Write data to Elasticsearch based on the official

The following code is taken from the official documentationElasticsearch based on the official

java, Elasticsearch 6.x ans above

  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. // Configuration for batch requests; the settings below cause the sink to commit immediately after receiving each element that would otherwise be cached
  44.         esSinkBuilder.setBulkFlushMaxActions(1);
  46. A RestClientFactory that provides custom configuration information for internally created REST clients
  47.         esSinkBuilder.setRestClientFactory(
  48.         restClientBuilder->{
  49.         restClientBuilder.setDefaultHeaders(...)
  50.         restClientBuilder.setMaxRetryTimeoutMillis(...)
  51.         restClientBuilder.setPathPrefix(...)
  52.         restClientBuilder.setHttpClientConfigCallback(...)
  53.         }
  54.         );
  56. // Finally, build and add the sink to the job pipeline
  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] =
  16. ...
  18. val httpHosts = new java.util.ArrayList[HttpHost]
  19. httpHosts.add(new HttpHost("127.0.0.1", 9200, "http"))
  20. httpHosts.add(new HttpHost("10.2.3.1", 9200, "http"))
  22. val esSinkBuilder = new ElasticsearchSink.Builder[String](
  23.   httpHosts,
  24.   new ElasticsearchSinkFunction[String] {
  25.     def process(element: String, ctx: RuntimeContext, indexer: RequestIndexer) {
  26.       val json = new java.util.HashMap[String, String]
  27.       json.put("data", element)
  29.       val rqst: IndexRequest = Requests.indexRequest
  30.         .index("my-index")
  31.         .`type`("my-type")
  32.         .source(json)
  34.       indexer.add(rqst)
  35.     }
  36.   }
  37. )
  39. // Configuration for batch requests; the settings below cause the sink to commit immediately after receiving each element that would otherwise be cached
  40. esSinkBuilder.setBulkFlushMaxActions(1)
  42. // A RestClientFactory that provides custom configuration information for internally created REST clients
  43. esSinkBuilder.setRestClientFactory(new RestClientFactory {
  44.   override def configureRestClientBuilder(restClientBuilder: RestClientBuilder): Unit = {
  45.     restClientBuilder.setDefaultHeaders(
  46.     ...)
  47.     restClientBuilder.setMaxRetryTimeoutMillis(
  48.     ...)
  49.     restClientBuilder.setPathPrefix(
  50.     ...)
  51.     restClientBuilder.setHttpClientConfigCallback(
  52.     ...)
  53.   }
  54. })
  56. // Finally, build and add the sink to the job pipeline
  57. input.addSink(esSinkBuilder.build)

The ElasticsearchSink created above is very inflexible to add parameters. StreamPark follows the concept of convention over configuration and automatic configuration. Users only need to configure es connection parameters and Flink operating parameters, and StreamPark will automatically assemble source and sink, which greatly simplifies development logic and improves development efficiency and maintainability.

Using Apache StreamPark™ writes to Elasticsearch

Please ensure that operation requests are sent to the Elasticsearch cluster at least once after enabling Flink checkpointing in ESSink.

1. 配置策略和连接信息

  1. #redis sink configure
  2. #  Required parameter, used by multiple nodes host1:port, host2:port,
  3. host: localhost:9200
  4. #  optional parameters
  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

Using StreamPark writes to 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)Define the writing rules for 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)define esSink and sink = data. 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.   }
  42.   def update6[T](stream: DataStream[T],
  43.                  suffix: String = "",
  44.                  restClientFactory: RestClientFactory = null,
  45.                  failureHandler: ActionRequestFailureHandler = new RetryRejectedExecutionFailureHandler)
  46.                 (f: T => UpdateRequest): DataStreamSink[T] = {
  48.     new ES6Sink(context, property, parallelism, name, uid).sink[T](stream, suffix, restClientFactory, failureHandler, f)
  49.   }
  52.   def delete6[T](stream: DataStream[T],
  53.                  suffix: String = "",
  54.                  restClientFactory: RestClientFactory = null,
  55.                  failureHandler: ActionRequestFailureHandler = new RetryRejectedExecutionFailureHandler)
  56.                 (f: T => DeleteRequest): DataStreamSink[T] = {
  58.     new ES6Sink(context, property, parallelism, name, uid).sink[T](stream, suffix, restClientFactory, failureHandler, f)
  59.   }
  61. }

When the Flink checkpoint is enabled, the Flink Elasticsearch Sink guarantees that operation requests are sent to the Elasticsearch cluster at least once. It does this by waiting for all pending operation requests in the BulkProcessor while checkpointing. This effectively guarantees that all requests will be successfully acknowledged by Elasticsearch before triggering the checkpoint and proceeding to process records sent to the sink. If the user wants to disable flushing, they can configure disableFlushOnCheckpoint to true to do so, which essentially means that the sink will no longer provide any reliable delivery guarantees, even if checkpointing of the job topology is enabled.

Other configuration

deal with failed Elasticsearch request

An Elasticsearch operation request may fail for a variety of reasons. You can specify the failure handling logic by implementing ActionRequestFailureHandler. See Official Documentation - Handling Failed Elasticsearch Requests

Configure the internal batch processor

The BulkProcessor inside es can further configure its behavior of how to refresh the cache operation request, see the official documentation for details - Configuring the Internal Bulk Processor

Apache StreamPark™ configuration

All other configurations must comply with the StreamPark configuration. For specific configurable items and the role of each parameter, please refer to the project configuration