当前位置:   article > 正文

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

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]]"},"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)

at com.esotericsoftware.kryo.serializers.FieldSerializer.read(FieldSerializer.java:528)

at com.esotericsoftware.kryo.Kryo.readObject(Kryo.java:679)

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

at com.esotericsoftware.kryo.serializers.FieldSerializer.read(FieldSerializer.java:528)

at com.esotericsoftware.kryo.Kryo.readObject(Kryo.java:679)

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

at com.esotericsoftware.kryo.serializers.FieldSerializer.read(FieldSerializer.java:528)

at com.esotericsoftware.kryo.Kryo.readObject(Kryo.java:679)

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

at com.esotericsoftware.kryo.serializers.FieldSerializer.read(FieldSerializer.java:528)

at com.esotericsoftware.kryo.Kryo.readObject(Kryo.java:679)

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

at com.esotericsoftware.kryo.serializers.FieldSerializer.read(FieldSerializer.java:528)

at com.esotericsoftware.kryo.Kryo.readObject(Kryo.java:657)

at org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer.copy(KryoSerializer.java:273)

at org.apache.flink.runtime.state.ArrayListSerializer.copy(ArrayListSerializer.java:75)

at org.apache.flink.runtime.state.PartitionableListState.<init>(PartitionableListState.java:64)

at org.apache.flink.runtime.state.PartitionableListState.deepCopy(PartitionableListState.java:76)

at org.apache.flink.runtime.state.DefaultOperatorStateBackendSnapshotStrategy.syncPrepareResources(DefaultOperatorStateBackendSnapshotStrategy.java:77)

at org.apache.flink.runtime.state.DefaultOperatorStateBackendSnapshotStrategy.syncPrepareResources(DefaultOperatorStateBackendSnapshotStrategy.java:36)

at org.apache.flink.runtime.state.SnapshotStrategyRunner.snapshot(SnapshotStrategyRunner.java:77)

at org.apache.flink.runtime.state.DefaultOperatorStateBackend.snapshot(DefaultOperatorStateBackend.java:230)

at org.apache.flink.streaming.api.operators.StreamOperatorStateHandler.snapshotState(StreamOperatorStateHandler.java:227)

... 35 more

Caused by: java.lang.UnsupportedOperationException

at java.util.Collections$UnmodifiableMap.put(Collections.java:1457)

at com.esotericsoftware.kryo.serializers.MapSerializer.read(MapSerializer.java:144)

at com.esotericsoftware.kryo.serializers.MapSerializer.read(MapSerializer.java:21)

at com.esotericsoftware.kryo.Kryo.readObject(Kryo.java:679)

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

... 58 more

解决方案

        1、sink 方案

        手写需要自己写一个sink,我自己从网上找了一个https://github.com/mtfelisb/flink-connector-elasticsearch 自己去照着写一个sink就可以了,需要补充的是对于一些超时问题,我们可以根据实际情况自己来设定。

        

        2、解决异常1

        因为这个问题不是因为flink导致,就是因为请求太频繁了,导致es队列超额了。所以我们可以控制一下请求的频率,比如 调高一下每次请求的条数,条数提高了相应请求次数也会降低。如果请求失败了,那么我们可以睡眠1秒,因为观察我们的es集群其实就是那么几秒内的请求太多,过了这段时间可能就下来了。我们停留一秒以后,再次发起请求就会成功。然后就是我们添加初始化状态,如果第二次请求还是等待超时异常,导致作业挂了。flink会自动重启,重启的时候我们需要从状态把上次请求的数据再次发送一下,避免数据丢失。

        3、解决异常2

        用es 8.x版本的时候,对于我们时间类型的字段类型没有办法支持,因为我们初始化客户端的时候用的是Jackson,对于java的时间类型是需要特殊处理一下的。异常其实已经告诉我们怎么处理了,添加好依赖,然后再初始化客户端的时候把时间模块注册进去就可以了。

        4、解决异常3

        这个异常,我也是找了一下网上的方案(kryo - Using an Collections$UnmodifiableCollection with Apache Flink - Stack Overflow)。首先这个问题是怎么产生的,因为我们写es的时候要用到script操作。当我们在script用到params的时候,就有了此问题。8.x的params是 Map<String,JsonData> 类型,而此类型在过程中序列不是Kryo,但是ES又是用的Kryo。我们又没有告诉他要用Kryo,所以就报异常了。我们在flink 环境初始化的时候指定一下就可以了。

  1. Class<?> unmodColl = Class.forName("java.util.Collections$UnmodifiableMap");
  2. env.getConfig().addDefaultKryoSerializer(unmodColl, UnmodifiableMapSerializer.class);

总结

        多干活才能发现这些问题,虽然有时候网上的解决方案和我们的不是一模一样,但是同样可以给我们带来新的思路。关键是解决问题的思维很关键。大胆尝试才能不断进步,舒适圈才会越来越大!!!

        如有说的不对的地方,还请指出,我们一起进步!

声明:本文内容由网友自发贡献,不代表【wpsshop博客】立场,版权归原作者所有,本站不承担相应法律责任。如您发现有侵权的内容,请联系我们。转载请注明出处:https://www.wpsshop.cn/w/Cpp五条/article/detail/606478
推荐阅读
相关标签
  

闽ICP备14008679号