searchusermenu
  • 发布文章
  • 消息中心
点赞
收藏
评论
分享
原创

Flink 读写 ElasticSearch 源码解析

2024-10-17 09:34:43
4
0

一、ElasticSearchSink介绍

在使用Flink进行数据的处理的时候,一个必要步骤就是需要将计算的结果进行存储或导出,Flink中这个过程称为Sink,官方我们提供了常用的几种Sink Connector,例如:

Apache Kafka

Elasticsearch

Elasticsearch 2x

Hadoop 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的空壳,定时执行的框架来实现了自己的一套失败重试机制。这一点很值得我们借鉴,很多知识点要学会活学活用,他山之石可以攻玉。

 

 

五、代码报错处理

1.类型错误

无法推断org.apache.flink.streaming.connectors.elasticsearch7.ElasticsearchSink.Builder<>的类型参数   

原因: 无法推断类型变量 T     (参数不匹配; java.util.List<java.net.InetSocketAddress>无法转换为java.util.List<org.apache.flink.elasticsearch7.shaded.org.apache.http.HttpHost>)

这个错误是因为在使用ElasticsearchSink.Builder时,传递的参数类型不正确。ElasticsearchSink.Builder需要接收org.apache.flink.elasticsearch7.shaded.org.apache.http.HttpHost类型的列表作为参数,而你传递的是java.net.InetSocketAddress类型的列表。

要解决这个问题,你需要将java.net.InetSocketAddress转换为org.apache.flink.elasticsearch7.shaded.org.apache.http.HttpHost。可以使用以下代码示例来实现转换:

List<InetSocketAddress> addresses = new ArrayList<>();// 添加InetSocketAddress到addresses列表中

 

List<HttpHost> httpHosts = addresses.stream()

    .map(address -> new HttpHost(address.getHostName(), address.getPort(), address.getScheme()))

    .collect(Collectors.toList());

 

ElasticsearchSink.Builder<T> esSinkBuilder = new ElasticsearchSink.Builder<>(httpHosts, new ElasticsearchSinkFunction<T>() {

    // ElasticsearchSinkFunction的实现

});

在上面的代码中,我们使用stream()map()方法将InetSocketAddress转换为HttpHost,然后使用collect()方法将其收集到一个新的List<HttpHost>中。然后,我们可以使用转换后的httpHosts列表来构建ElasticsearchSink.Builder

请确保在使用ElasticsearchSink.Builder时传递正确的参数类型,以解决这个问题。

 

2.类调用问题

在自定义异常处理时,不要调用import org.apache.flink.elasticsearch7.shaded.org.apache.http.util.ExceptionUtils;而是需要调用import org.apache.flink.util.ExceptionUtils;

否则会报相关的类的方法找不到的错误。前者已经废弃了。

ps:很多方法无法调用的错误,都需要检查一下调用的类名所在的包是不是导错了)

 

六、源码走读

1. ElasticsearchSink 类

该类继承自ElasticsearchSinkBase类,类型为泛型,使用时需要指定类型

 一般性的,我们使用 ElasticsearchSink.Builder 创建 ElasticsearchSinkBuilder类同样使用泛型指定类型。

该类有很多set方法用于设置ES Sink的参数。 

而该类的构造函数public Builder()会创建一个新的{@code ElasticsearchSink},它使用{@link RestHighLevelClient}连接到集群。

需要传入两个参数:

@param httpHosts {@link HttpHost}列表,连接到{@link RestHighLevelClient}的主机。

@param elasticsearchSinkFunction 用于从传入元素生成多个{@link ActionRequest}的函数。

0条评论
0 / 1000
Sirius.
5文章数
0粉丝数
Sirius.
5 文章 | 0 粉丝
原创

Flink 读写 ElasticSearch 源码解析

2024-10-17 09:34:43
4
0

一、ElasticSearchSink介绍

在使用Flink进行数据的处理的时候,一个必要步骤就是需要将计算的结果进行存储或导出,Flink中这个过程称为Sink,官方我们提供了常用的几种Sink Connector,例如:

Apache Kafka

Elasticsearch

Elasticsearch 2x

Hadoop 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的空壳,定时执行的框架来实现了自己的一套失败重试机制。这一点很值得我们借鉴,很多知识点要学会活学活用,他山之石可以攻玉。

 

 

五、代码报错处理

1.类型错误

无法推断org.apache.flink.streaming.connectors.elasticsearch7.ElasticsearchSink.Builder<>的类型参数   

原因: 无法推断类型变量 T     (参数不匹配; java.util.List<java.net.InetSocketAddress>无法转换为java.util.List<org.apache.flink.elasticsearch7.shaded.org.apache.http.HttpHost>)

这个错误是因为在使用ElasticsearchSink.Builder时,传递的参数类型不正确。ElasticsearchSink.Builder需要接收org.apache.flink.elasticsearch7.shaded.org.apache.http.HttpHost类型的列表作为参数,而你传递的是java.net.InetSocketAddress类型的列表。

要解决这个问题,你需要将java.net.InetSocketAddress转换为org.apache.flink.elasticsearch7.shaded.org.apache.http.HttpHost。可以使用以下代码示例来实现转换:

List<InetSocketAddress> addresses = new ArrayList<>();// 添加InetSocketAddress到addresses列表中

 

List<HttpHost> httpHosts = addresses.stream()

    .map(address -> new HttpHost(address.getHostName(), address.getPort(), address.getScheme()))

    .collect(Collectors.toList());

 

ElasticsearchSink.Builder<T> esSinkBuilder = new ElasticsearchSink.Builder<>(httpHosts, new ElasticsearchSinkFunction<T>() {

    // ElasticsearchSinkFunction的实现

});

在上面的代码中,我们使用stream()map()方法将InetSocketAddress转换为HttpHost,然后使用collect()方法将其收集到一个新的List<HttpHost>中。然后,我们可以使用转换后的httpHosts列表来构建ElasticsearchSink.Builder

请确保在使用ElasticsearchSink.Builder时传递正确的参数类型,以解决这个问题。

 

2.类调用问题

在自定义异常处理时,不要调用import org.apache.flink.elasticsearch7.shaded.org.apache.http.util.ExceptionUtils;而是需要调用import org.apache.flink.util.ExceptionUtils;

否则会报相关的类的方法找不到的错误。前者已经废弃了。

ps:很多方法无法调用的错误,都需要检查一下调用的类名所在的包是不是导错了)

 

六、源码走读

1. ElasticsearchSink 类

该类继承自ElasticsearchSinkBase类,类型为泛型,使用时需要指定类型

 一般性的,我们使用 ElasticsearchSink.Builder 创建 ElasticsearchSinkBuilder类同样使用泛型指定类型。

该类有很多set方法用于设置ES Sink的参数。 

而该类的构造函数public Builder()会创建一个新的{@code ElasticsearchSink},它使用{@link RestHighLevelClient}连接到集群。

需要传入两个参数:

@param httpHosts {@link HttpHost}列表,连接到{@link RestHighLevelClient}的主机。

@param elasticsearchSinkFunction 用于从传入元素生成多个{@link ActionRequest}的函数。

文章来自个人专栏
Sirius
5 文章 | 1 订阅
0条评论
0 / 1000
请输入你的评论
0
0