线上业务反应使用 Flink 消费上游 kafka topic 里的轨迹数据出现 backpressure，数据积压严重。单次 bulk 的写入量为：3000/50mb/30s，并行度为 48。针对该问题，为了避免影响线上业务申请了一个与线上集群配置相同的 ES 集群。本着复现问题进行优化就能解决的思路进行调优测试。

• Elasticsearch 2.3.3

• Flink 1.6.3

• flink-connector-elasticsearch 2_2.11

• 八台 SSD，56 核 ：3 主 5 从

• 从压测结果来看，集群层面的平均写入性能大概在每秒 10 w+ 的 doc。

• 配置文件

1config.put("cluster.name", ConfigUtil.getString(ES_CLUSTER_NAME, "flinktest"));	
2config.put("bulk.flush.max.actions", ConfigUtil.getString(ES_BULK_FLUSH_MAX_ACTIONS, "3000"));	
3config.put("bulk.flush.max.size.mb", ConfigUtil.getString(ES_BULK_FLUSH_MAX_SIZE_MB, "50"));	
4config.put("bulk.flush.interval.ms", ConfigUtil.getString(ES_BULK_FLUSH_INTERVAL, "3000"));

• 执行代码片段

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();	
 initEnv(env);	
 Properties properties = ConfigUtil.getProperties(CONFIG_FILE_PATH);	
 //从kafka中获取轨迹数据	
 FlinkKafkaConsumer010<String> flinkKafkaConsumer010 =	
     new FlinkKafkaConsumer010<>(properties.getProperty("topic.name"), new SimpleStringSchema(), properties);	
 //从checkpoint最新处消费	
 flinkKafkaConsumer010.setStartFromLatest();	
 DataStreamSource<String> streamSource = env.addSource(flinkKafkaConsumer010);	
10//Sink2ES	
streamSource.map(s -> JSONObject.parseObject(s, Trajectory.class))	
    .addSink(EsSinkFactory.createSinkFunction(new TrajectoryDetailEsSinkFunction())).name("esSink");	
env.execute("flinktest");

• 运行时配置

任务容器数为 24 个 container，一共 48 个并发。savepoint 为 15 分钟：

• 运行现象

（1）source 和 Map 算子均出现较高的反压

（2）ES 集群层面，目标索引写入速度写入陡降

平均 QPS 为：12 k 左右。

（3）对比取消 sink 算子后的 QPS

平均QPS为：116 k 左右。

有无sink参照实验的结论：

取消 sink 2 ES 的操作后，QPS 达到 110 k,是之前 QPS 的十倍。由此可以基本判定: ES 集群写性能导致的上游反压

• 索引字段类型调整

bulk 失败的原因是由于集群 dynamic mapping 自动监测，部分字段格式被识别为日期格式而遇到空字符串无法解析报错。

解决方案：关闭索引自动检测。

效果: ES 集群写入性能明显提高但 Flink operator 依然存在反压：

• 降低副本数

• 提高 refresh_interval

针对这种 ToB、日志型、实时性要求不高的场景，我们不需要查询的实时性，通过加大甚至关闭 refresh_interval 的参数提高写入性能。

• 检查集群各个节点 CPU 核数

在 Flink 执行时，通过 Grafana 观测各个节点 CPU 使用率以及通过 Linux 命令查看各个节点 CPU 核数。发现 CPU 使用率高的节点 CPU 核数比其余节点少。为了排除这个短板效应，我们将在这个节点中的索引 shard 移动到 CPU 核数多的节点。

curl -XPOST {集群地址}/_cluster/reroute  -d'{"commands":[{"move":{"index":"{索引名称}","shard":5,"from_node":"源node名称","to_node":"目标node名称"}}]}' -H "Content-Type:application/json"

以上优化的效果:

经过以上的优化，我们发现写入性能提升有限。因此，需要深入查看写入的瓶颈点。

• 在 CPU 使用率高的节点使用 Arthas 观察线程

• 打印阻塞的线程堆栈

 "elasticsearch[ES-077-079][bulk][T#3]" Id=247 WAITING on java.util.concurrent.LinkedTransferQueue@369223fa	
   at sun.misc.Unsafe.park(Native Method)	
     -  waiting on java.util.concurrent.LinkedTransferQueue@369223fa	
    at java.util.concurrent.locks.LockSupport.park(LockSupport.java:175)	
    at java.util.concurrent.LinkedTransferQueue.awaitMatch(LinkedTransferQueue.java:737)	
    at java.util.concurrent.LinkedTransferQueue.xfer(LinkedTransferQueue.java:647)	
    at java.util.concurrent.LinkedTransferQueue.take(LinkedTransferQueue.java:1269)	
    at org.elasticsearch.common.util.concurrent.SizeBlockingQueue.take(SizeBlockingQueue.java:161)	
    at java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:1067)	
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1127)	
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)	
    at java.lang.Thread.run(Thread.java:745)

从上面的线程堆栈我们可以看出线程处于等待状态。

关于这个问题的讨论详情查看 https://discuss.elastic.co/t/thread-selection-and-locking/26051/3，这个 issue 讨论大致意思是：节点数不够，需要增加节点。于是我们又增加节点并通过设置索引级别的 total_shards_per_node 参数将索引 shard 的写入平均到各个节点上。

• 线程队列优化

ES 是将不同种类的操作(index、search…)交由不同的线程池执行，主要的线程池有三：index、search 和 bulk thread_pool。线程池队列长度配置按照官网默认值，我觉得增加队列长度而集群本身没有很高的处理能力线程还是会 await（事实上实验结果也是如此在此不必赘述），因为实验节点机器是 56 核，对照官网：

因此修改 size 数值为 56。

经过以上的优化，我们发现在 kafka 中的 topic 积压有明显变少的趋势：

• index buffer size 的优化

参照官网：

• translog 优化：

索引写入 ES 的基本流程是：

• 数据写入 buffer 缓冲和 translog； 

• 每秒 buffer 的数据生成 segment 并进入内存，此时 segment 被打开并供 search 使用查询； 

• buffer 清空并重复上述步骤 ；

• buffer 不断添加、清空 translog 不断累加，当达到某些条件触发 commit 操作，刷到磁盘；

ES 默认的刷盘操作为 Request 但容易部分操作比较耗时，在日志型集群、允许数据在刷盘过程中少量丢失可以改成异步 async。

另外一次 commit 操作是在 translog 达到某个阈值执行的，可以把大小（flush_threshold_size ）调大，刷新间隔调大。

index.translog.durability : async	
index.translog.flush_threshold_size : 1gb	
index.translog.sync_interval : 30s

效果：

• Flink 反压从打满 100% 降到 40%（output buffer usage）：

• kafka 消费组里的积压明显减少：

当 ES 写入性能遇到瓶颈时，我总结的思路应该是这样：

• 看日志，是否有字段类型不匹配，是否有脏数据。

• 看 CPU 使用情况，集群是否异构

• 客户端是怎样的配置？使用的 bulk 还是单条插入

• 查看线程堆栈，查看耗时最久的方法调用

• 确定集群类型：ToB 还是 ToC，是否允许有少量数据丢失？

• 针对 ToB 等实时性不高的集群减少副本增加刷新时间

• index buffer 优化 translog 优化，滚动重启集群