《從0到1學習Flink》—— Flink 寫入數據到 ElasticSearch

2019 年 10 月 4 日
筆記

前言

前面 FLink 的文章中我們已經介紹了說 Flink 已經有很多自帶的 Connector。

1、《從0到1學習Flink》—— Data Source 介紹

2、《從0到1學習Flink》—— Data Sink 介紹

其中包括了 Source 和 Sink 的，後面我也講了下如何自定義自己的 Source 和 Sink。

那麼今天要做的事情是啥呢？就是介紹一下 Flink 自帶的 ElasticSearch Connector，我們今天就用他來做 Sink，將 Kafka 中的數據經過 Flink 處理後然後存儲到 ElasticSearch。

準備

安裝 ElasticSearch，這裡就忽略，自己找我以前的文章，建議安裝 ElasticSearch 6.0 版本以上的，畢竟要跟上時代的節奏。

下面就講解一下生產環境中如何使用 Elasticsearch Sink 以及一些注意點，及其內部實現機制。

Elasticsearch Sink

添加依賴

<dependency>      <groupId>org.apache.flink</groupId>      <artifactId>flink-connector-elasticsearch6_${scala.binary.version}</artifactId>      <version>${flink.version}</version>  </dependency>

上面這依賴版本號請自己根據使用的版本對應改變下。

下面所有的程式碼都沒有把 import 引入到這裡來，如果需要查看更詳細的程式碼，請查看我的 GitHub 倉庫地址：

https://github.com/zhisheng17/flink-learning/tree/master/flink-learning-connectors/flink-learning-connectors-es6

這個 module 含有本文的所有程式碼實現，當然越寫到後面自己可能會做一些抽象，所以如果有程式碼改變很正常，請直接查看全部項目程式碼。

ElasticSearchSinkUtil 工具類

這個工具類是自己封裝的，getEsAddresses 方法將傳入的配置文件 es 地址解析出來，可以是域名方式，也可以是 ip + port 形式。addSink 方法是利用了 Flink 自帶的 ElasticsearchSink 來封裝了一層，傳入了一些必要的調優參數和 es 配置參數，下面文章還會再講些其他的配置。

ElasticSearchSinkUtil.java

public class ElasticSearchSinkUtil {        /**       * es sink       *       * @param hosts es hosts       * @param bulkFlushMaxActions bulk flush size       * @param parallelism 並行數       * @param data 數據       * @param func       * @param <T>       */      public static <T> void addSink(List<HttpHost> hosts, int bulkFlushMaxActions, int parallelism,                                     SingleOutputStreamOperator<T> data, ElasticsearchSinkFunction<T> func) {          ElasticsearchSink.Builder<T> esSinkBuilder = new ElasticsearchSink.Builder<>(hosts, func);          esSinkBuilder.setBulkFlushMaxActions(bulkFlushMaxActions);          data.addSink(esSinkBuilder.build()).setParallelism(parallelism);      }        /**       * 解析配置文件的 es hosts       *       * @param hosts       * @return       * @throws MalformedURLException       */      public static List<HttpHost> getEsAddresses(String hosts) throws MalformedURLException {          String[] hostList = hosts.split(",");          List<HttpHost> addresses = new ArrayList<>();          for (String host : hostList) {              if (host.startsWith("http")) {                  URL url = new URL(host);                  addresses.add(new HttpHost(url.getHost(), url.getPort()));              } else {                  String[] parts = host.split(":", 2);                  if (parts.length > 1) {                      addresses.add(new HttpHost(parts[0], Integer.parseInt(parts[1])));                  } else {                      throw new MalformedURLException("invalid elasticsearch hosts format");                  }              }          }          return addresses;      }  }

Main 啟動類

Main.java

public class Main {      public static void main(String[] args) throws Exception {          //獲取所有參數          final ParameterTool parameterTool = ExecutionEnvUtil.createParameterTool(args);          //準備好環境          StreamExecutionEnvironment env = ExecutionEnvUtil.prepare(parameterTool);          //從kafka讀取數據          DataStreamSource<Metrics> data = KafkaConfigUtil.buildSource(env);            //從配置文件中讀取 es 的地址          List<HttpHost> esAddresses = ElasticSearchSinkUtil.getEsAddresses(parameterTool.get(ELASTICSEARCH_HOSTS));          //從配置文件中讀取 bulk flush size，代表一次批處理的數量，這個可是性能調優參數，特別提醒          int bulkSize = parameterTool.getInt(ELASTICSEARCH_BULK_FLUSH_MAX_ACTIONS, 40);          //從配置文件中讀取並行 sink 數，這個也是性能調優參數，特別提醒，這樣才能夠更快的消費，防止 kafka 數據堆積          int sinkParallelism = parameterTool.getInt(STREAM_SINK_PARALLELISM, 5);            //自己再自帶的 es sink 上一層封裝了下          ElasticSearchSinkUtil.addSink(esAddresses, bulkSize, sinkParallelism, data,                  (Metrics metric, RuntimeContext runtimeContext, RequestIndexer requestIndexer) -> {                      requestIndexer.add(Requests.indexRequest()                              .index(ZHISHENG + "_" + metric.getName())  //es 索引名                              .type(ZHISHENG) //es type                              .source(GsonUtil.toJSONBytes(metric), XContentType.JSON));                  });          env.execute("flink learning connectors es6");      }  }

配置文件

配置都支援集群模式填寫，注意用 , 分隔！

kafka.brokers=localhost:9092  kafka.group.id=zhisheng-metrics-group-test  kafka.zookeeper.connect=localhost:2181  metrics.topic=zhisheng-metrics  stream.parallelism=5  stream.checkpoint.interval=1000  stream.checkpoint.enable=false  elasticsearch.hosts=localhost:9200  elasticsearch.bulk.flush.max.actions=40  stream.sink.parallelism=5

運行結果

執行 Main 類的 main 方法，我們的程式是只列印 flink 的日誌，沒有列印存入的日誌（因為我們這裡沒有打日誌）：

所以看起來不知道我們的 sink 是否有用，數據是否從 kafka 讀取出來後存入到 es 了。

你可以查看下本地起的 es 終端或者伺服器的 es 日誌就可以看到效果了。

es 日誌如下：

上圖是我本地 Mac 電腦終端的 es 日誌，可以看到我們的索引了。

如果還不放心，你也可以在你的電腦裝個 kibana，然後更加的直觀查看下 es 的索引情況（或者直接敲 es 的命令）

我們用 kibana 查看存入 es 的索引如下：

程式執行了一會，存入 es 的數據量就很大了。

擴展配置

上面程式碼已經可以實現你的大部分場景了，但是如果你的業務場景需要保證數據的完整性（不能出現丟數據的情況），那麼就需要添加一些重試策略，因為在我們的生產環境中，很有可能會因為某些組件不穩定性導致各種問題，所以這裡我們就要在數據存入失敗的時候做重試操作，這裡 flink 自帶的 es sink 就支援了，常用的失敗重試配置有:

bulk.flush.backoff.enable 用來表示是否開啟重試機制    bulk.flush.backoff.type 重試策略，有兩種：EXPONENTIAL 指數型（表示多次重試之間的時間間隔按照指數方式進行增長）、CONSTANT 常數型（表示多次重試之間的時間間隔為固定常數）    bulk.flush.backoff.delay 進行重試的時間間隔    bulk.flush.backoff.retries 失敗重試的次數    bulk.flush.max.actions: 批量寫入時的最大寫入條數    bulk.flush.max.size.mb: 批量寫入時的最大數據量    bulk.flush.interval.ms: 批量寫入的時間間隔，配置後則會按照該時間間隔嚴格執行，無視上面的兩個批量寫入配置

看下啦，就是如下這些配置了，如果你需要的話，可以在這個地方配置擴充了。

FailureHandler 失敗處理器

寫入 ES 的時候會有這些情況會導致寫入 ES 失敗：

1、ES 集群隊列滿了，報如下錯誤

[I/O dispatcher 13] ERROR o.a.f.s.c.e.ElasticsearchSinkBase - Failed Elasticsearch item request: ElasticsearchException[Elasticsearch exception [type=es_rejected_execution_exception, reason=rejected execution of org.elasticsearch.transport.TransportService$7@566c9379 on EsThreadPoolExecutor[name = node-1/write, queue capacity = 200, org.elasticsearch.common.util.concurrent.EsThreadPoolExecutor@f00b373[Running, pool size = 4, active threads = 4, queued tasks = 200, completed tasks = 6277]]]]

是這樣的，我電腦安裝的 es 隊列容量默認應該是 200，我沒有修改過。我這裡如果配置的 bulk flush size * 並發 sink 數量這個值如果大於這個 queue capacity ，那麼就很容易導致出現這種因為 es 隊列滿了而寫入失敗。

當然這裡你也可以通過調大點 es 的隊列。參考：https://www.elastic.co/guide/en/elasticsearch/reference/current/modules-threadpool.html

2、ES 集群某個節點掛了

這個就不用說了，肯定寫入失敗的。跟過源碼可以發現 RestClient 類里的 performRequestAsync 方法一開始會隨機的從集群中的某個節點進行寫入數據，如果這台機器掉線，會進行重試在其他的機器上寫入，那麼當時寫入的這台機器的請求就需要進行失敗重試，否則就會把數據丟失！

3、ES 集群某個節點的磁碟滿了

這裡說的磁碟滿了，並不是磁碟真的就沒有一點剩餘空間的，是 es 會在寫入的時候檢查磁碟的使用情況，在 85% 的時候會列印日誌警告。

這裡我看了下源碼如下圖：

如果你想繼續讓 es 寫入的話就需要去重新配一下 es 讓它繼續寫入，或者你也可以清空些不必要的數據騰出磁碟空間來。

解決方法

DataStream<String> input = ...;    input.addSink(new ElasticsearchSink<>(      config, transportAddresses,      new ElasticsearchSinkFunction<String>() {...},      new ActionRequestFailureHandler() {          @Override          void onFailure(ActionRequest action,                  Throwable failure,                  int restStatusCode,                  RequestIndexer indexer) throw Throwable {                if (ExceptionUtils.containsThrowable(failure, EsRejectedExecutionException.class)) {                  // full queue; re-add document for indexing                  indexer.add(action);              } else if (ExceptionUtils.containsThrowable(failure, ElasticsearchParseException.class)) {                  // malformed document; simply drop request without failing sink              } else {                  // for all other failures, fail the sink                  // here the failure is simply rethrown, but users can also choose to throw custom exceptions                  throw failure;              }          }  }));

如果僅僅只是想做失敗重試，也可以直接使用官方提供的默認的 RetryRejectedExecutionFailureHandler ，該處理器會對 EsRejectedExecutionException 導致到失敗寫入做重試處理。如果你沒有設置失敗處理器(failure handler)，那麼就會使用默認的 NoOpFailureHandler 來簡單處理所有的異常。

總結

本文寫了 Flink connector es，將 Kafka 中的數據讀取並存儲到 ElasticSearch 中，文中講了如何封裝自帶的 sink，然後一些擴展配置以及 FailureHandler 情況下要怎麼處理。（這個問題可是線上很容易遇到的）

關注我

轉載請務必註明原創地址為：http://www.54tianzhisheng.cn/2018/12/30/Flink-ElasticSearch-Sink/