一、ElasticSearchSink 介绍
在使用 Flink 进行数据的处理的时候，一个必要步骤就是需要将计算的结果进行存储或导出，Flink 中这个过程称为 Sink，官方我们提供了常用的几种 Sink Connector，例如:
Apache KafkaElasticsearchElasticsearch 2xHadoop FileSystem…
复制代码
这篇就选取其中一个常用的 ElasticsearchSink 来进行介绍，并讲解一下生产环境中使用时的一些注意点，以及其内部实现机制。
二、使用方式
1. 添加 pom 依赖
<dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-connector-elasticsearch2_2.10</artifactId> <version>1.3.1</version></dependency>
复制代码
根据自己所用的 filnk 版本以及 es 版本对上面的版本号进行调整
2. 实现对应代码
DataStream<String> input = ...;
Map<String, String> config = new HashMap<>();config.put("cluster.name", "my-cluster-name");//该配置表示批量写入ES时的记录条数config.put("bulk.flush.max.actions", "1");
List<InetSocketAddress> transportAddresses = new ArrayList<>();transportAddresses.add(new InetSocketAddress(InetAddress.getByName("127.0.0.1"), 9300));transportAddresses.add(new InetSocketAddress(InetAddress.getByName("10.2.3.1"), 9300));
input.addSink(new ElasticsearchSink<>(config, transportAddresses, new ElasticsearchSinkFunction<String>() { public IndexRequest createIndexRequest(String element) { Map<String, String> json = new HashMap<>(); //将需要写入ES的字段依次添加到Map当中 json.put("data", element);
return Requests.indexRequest() .index("my-index") .type("my-type") .source(json); }
@Override public void process(String element, RuntimeContext ctx, RequestIndexer indexer) { indexer.add(createIndexRequest(element)); }}));
复制代码
扩展配置
经过上面的代码已经实现了一个基础版的 EsSink，但是上述代码当 ES 集群出现波动的时候，由于不具备重试机制则有可能出现丢数据的情况。生产环境中为了实现数据完整性，我们需要添加一些失败重试配置，来实现写入失败情况下的容错处理，常用的失败重试配置有:
//1、用来表示是否开启重试机制config.put("bulk.flush.backoff.enable", "true");//2、重试策略，又可以分为以下两种类型 //a、指数型，表示多次重试之间的时间间隔按照指数方式进行增长。eg:2 -> 4 -> 8 ... config.put("bulk.flush.backoff.type", "EXPONENTIAL"); //b、常数型，表示多次重试之间的时间间隔为固定常数。eg:2 -> 2 -> 2 ... config.put("bulk.flush.backoff.type", "CONSTANT");//3、进行重试的时间间隔。对于指数型则表示起始的基数config.put("bulk.flush.backoff.delay", "2");//4、失败重试的次数config.put("bulk.flush.backoff.retries", "3");
复制代码
其他的一些配置:
bulk.flush.max.actions: 批量写入时的最大写入条数 bulk.flush.max.size.mb: 批量写入时的最大数据量 bulk.flush.interval.ms: 批量写入的时间间隔，配置后则会按照该时间间隔严格执行，无视上面的两个批量写入配置
三、失败处理器
写入 ES 的时候很多时候由于 ES 集群队列满了，或者节点挂掉，经常会导致写入操作执行失败。考虑到这样的失败写入场景，EsSink 为用户提供了失败处理器机制，创建 Sink 对象的时候，同时可以传入一个失败处理器，一旦出现写入失败的情况则会回调所传入的处理器用于错误恢复。具体的用法为:
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)) { // 将失败请求继续加入队列，后续进行重试写入 indexer.add(action); } else if (ExceptionUtils.containsThrowable(failure, ElasticsearchParseException.class)) { // 添加自定义的处理逻辑 } else { throw failure; } }}));
复制代码
如果仅仅只是想做失败重试，也可以直接使用官方提供的默认的 RetryRejectedExecutionFailureHandler，该处理器会对 EsRejectedExecutionException 导致到失败写入做重试处理。
四、其他注意点
1. EsSink 代码块不能使用 try-catch-Exception 来捕捉
之前在使用 EsSink 的时候，为了防止某次写入失败造成程序中断，对 ElasticsearchSinkFunction 的 process() 方法使用 try-catch-exception 语句块进行了捕捉，但实际运行的时候发现程序跑着跑着还是被一个 EsRejectedException 异常中断掉了。让人奇怪的是明明对异常进行了捕捉，为什么这个异常还是能够抛出来，下来通过查看源码发现，如果在初始化 EsSink 对象的时候没有传入 ActionRequestFailureHandler 则会使用默认的 ActionRequestFailureHandler ，这个处理器的源码如下:
public class NoOpFailureHandler implements ActionRequestFailureHandler {
private static final long serialVersionUID = 737941343410827885L;
@Override public void onFailure(ActionRequest action, Throwable failure, int restStatusCode, RequestIndexer indexer) throws Throwable { // 这里抛出的是一个throwable throw failure; }
复制代码
可以看到，在发生异常的时候，默认的处理器会将异常包装成一个 Throw 对象抛出，这就是直接使用 try-Exception 无法捕捉到的原因。
解决方法:
实现自己的失败处理器消化掉异常
使用 throw 来捕捉异常
该问题一定要重点注意，负责会导致实时任务终止掉！
2. 失败重试机制依赖于 checkpoint
如果想要使用 EsSink 的失败重试机制，则需要通过 env.enableCheckpoint() 方法来开启 Flink 任务对 checkpoint 的支持，如果没有开启 checkpoint 机制的话，则失败重试策略是无法生效的。这个是通过跟踪 ElasticsearchSinkBase 类源码的时候发现的，核心的代码如下:
@Override public void initializeState(FunctionInitializationContext context) throws Exception { // no initialization needed }
@Override public void snapshotState(FunctionSnapshotContext context) throws Exception { checkErrorAndRethrow(); //如果没有开启checkPoint机制，则该变量为false，也就导致下面的flush重试代码不会执行到 if (flushOnCheckpoint) { do { //失败重试的时机是发生在程序在打checkpoint的时候 bulkProcessor.flush(); checkErrorAndRethrow(); } while (numPendingRequests.get() != 0); } }
复制代码
3. 总结
可以通过第二点贴出的源码发现，虽然 EsSink 实现了 CheckpointedFunction 接口,并且重写了 checkPoint 的相关方法，但其并没有墨守成规的利用 checkpoint 定义的那样利用 State 机制用于故障恢复。而是利用了 checkpoint 的空壳，定时执行的框架来实现了自己的一套失败重试机制。这一点很值得我们借鉴，很多知识点要学会活学活用，他山之石可以攻玉。
评论