Flink1.14自定义ES8.Xsink_flink自定义es sink

作者：Cpp五条 | 2024-05-22 05:01:22

踩

flink自定义es sink

环境

flink 1.14.4
ES 8.6.2
Linux version 3.10.0-862.14.4.el7.x86_64 (mockbuild@kbuilder.bsys.centos.org)
java version "1.8.0_131"

问题描述

flink1.14.4 写入 ES 8.6.2 集群，如果集群在正常的情况本来是没有问题的，但是集群在写的过程中，如果ES的线程池队列如果比较繁忙，超过队列长度的情况。用我之前的flink sink（社区的外加自己改了一点）是有一定的问题，表现的现象就是数据丢失，当然这不是说flink丢数据而是因为ES集群的问题，导致写数据一直超时，超过了重试次数。报出来的异常没有明显的提示。所以只能自己自定义个对应ES8.6.2的客户端来进行优化。

异常描述

1、es线程池队列问题

{"error":{"root_cause":[{"type":"es_rejected_execution_exception","reason":"rejected execution of org.elasticsearch.action.bulk.TransportBulkAction$1/org.elasticsearch.action.ActionListener$RunBeforeActionListener/org.elasticsearch.tasks.TaskManager$1{SafelyWrappedActionListener[listener=org.elasticsearch.rest.action.RestStatusToXContentListener@55a0fc06]}{Task{id=41565354241, type='transport', action='indices:data/write/bulk', description='requests[500], indices[xxxx]', parentTask=unset, startTime=1700033142640, startTimeNanos=71265745979184895}}/org.elasticsearch.action.bulk.TransportBulkAction

Lambda$7442/0x000000080221f978@1bf83811 on EsThreadPoolExecutor[name =xxx/write, queue capacity = 10000, org.elasticsearch.common.util.concurrent.EsThreadPoolExecutor@c7443f[Running, pool size = 16, active threads = 16, queued tasks = 10979, completed tasks = 5335112349]]"}],"type":"es_rejected_execution_exception","reason":"rejected execution of org.elasticsearch.action.bulk.TransportBulkAction$1/org.elasticsearch.action.ActionListener$RunBeforeActionListener/org.elasticsearch.tasks.TaskManager$1{SafelyWrappedActionListener[listener=org.elasticsearch.rest.action.RestStatusToXContentListener@55a0fc06]}{Task{id=41565354241, type='transport', action='indices:data/write/bulk', description='requests[500], indices[xxx]', parentTask=unset, startTime=1700033142640, startTimeNanos=71265745979184895}}/org.elasticsearch.action.bulk.TransportBulkAction

$Lambda$7442/0x000000080221f978@1bf83811 on EsThreadPoolExecutor[name =xxx/write, queue capacity = 10000, org.elasticsearch.common.util.concurrent.EsThreadPoolExecutor@c7443f[Running, pool size = 16, active threads = 16, queued tasks = 10979, completed tasks = 5335112349]]"}],"type":"es_rejected_execution_exception","reason":"rejected execution of org.elasticsearch.action.bulk.TransportBulkAction$1/org.elasticsearch.action.ActionListener$RunBeforeActionListener/org.elasticsearch.tasks.TaskManager$1{SafelyWrappedActionListener[listener=org.elasticsearch.rest.action.RestStatusToXContentListener@55a0fc06]}{Task{id=41565354241, type='transport', action='indices:data/write/bulk', description='requests[500], indices[xxx]', parentTask=unset, startTime=1700033142640, startTimeNanos=71265745979184895}}/org.elasticsearch.action.bulk.TransportBulkAction$ Lambda$7442/0x000000080221f978@1bf83811 on EsThreadPoolExecutor[name = xxx/write, queue capacity = 10000, org.elasticsearch.common.util.concurrent.EsThreadPoolExecutor@c7443f[Running, pool size = 16, active threads = 16, queued tasks = 10979, completed tasks = 5335112349]]"},"status":429}

2、时间类型不支持

ava 8 date/time type `java.time.LocalDate` not supported by default: add Module "com.fasterxml.jackson.datatype:jackson-datatype-jsr310" to enable handling

3、序列化不支持问题

Caused by: com.esotericsoftware.kryo.KryoException: java.lang.UnsupportedOperationException

Serialization trace:

params (co.elastic.clients.elasticsearch._types.InlineScript)

_value (co.elastic.clients.elasticsearch._types.Script)

script (co.elastic.clients.elasticsearch.core.bulk.UpdateAction)

action (co.elastic.clients.elasticsearch.core.bulk.UpdateOperation)

_value (co.elastic.clients.elasticsearch.core.bulk.BulkOperation)

at com.esotericsoftware.kryo.serializers.ObjectField.read(ObjectField.java:125)