Flink-异步算子AsyncFunction利用

配景

        在Flink使命中必要将盘算结果数据先存入HBase中（Phoenix），在保证HBase写入成功后再转发到Kafka中。在数据量较大的情况下由于串行写入HBase会出现严峻反压，造成整体数据盘算链路的整体数据产出效率大大低落。
目标

        利用异步算子与外部系统交互，提高吞吐量，低落由于写入数据库造成的耽误。
描述

        在本地测试时验证出，一条数据写入的HBase的哀求完备耗时在40ms左右（仅作为参考，不怜悯况耗时差异较大），但是在Flink使命中上游算子的耗时基本都在几毫秒内，因此写入HBase是整体链路的显着短板。在一个写入HBase的Task中，由于写入哀求是同步执行的，在大部分的情况下等待哀求相应占据了该算子的大部分时间，因此通过并发的实现处理多个哀求和接收多个响应，将等待的时间分摊到每个哀求。
            注：仅仅提高Function的并行度（parallelism）在有些情况下也可以提升吞吐量，但是如许做通常会导致非常高的资源斲丧：更多的并行 Function 实例意味着更多的 Task、更多的线程、更多的 Flink 内部网络毗连、 更多的与数据库的网络毗连、更多的缓冲和更多程序内部协调的开销。
                                                                                                  --Apache Flink Documentation
  
两个参数控制异步操作：
Timeout： 超时参数界说了异步哀求发出多久后未得到响应即被认定为失败。 它可以防止一直等待得不到响应的哀求。
Capacity： 容量参数界说了可以同时进行的异步哀求数。 即使异步 I/O 通常带来更高的吞吐量，执行异步 I/O 操作的算子仍然可能成为流处理的瓶颈。 限定并发哀求的数量可以确保算子不会连续累积待处理的哀求进而造成积压，而是在容量耗尽时触发反压。
留意点：

• 在利用与外部数据库交互的客户端时，需利用支持异步操作的客户端，若没有则可以利用线程池实现异步提交的哀求，若利用同步的客户端仍会壅闭当前异步算子。
  
• 在利用异步算子时，Capacity参数、线程池最大毗连数、数据库毗连池最大毗连数要合理设置，克制由于某个参数设置的过小使哀求的并发数受限，无法达到期望并发。
  
•  官方文档的中已对 超时处理、结果的顺序、事件时间、容错保证做了明白的解释，放在文档最后便于查看。
  

代码

1. package app;
3. import com.alibaba.fastjson.JSONObject;
4. import functions.MyAsyncFunction;
5. import org.apache.flink.api.common.functions.MapFunction;
6. import org.apache.flink.streaming.api.datastream.AsyncDataStream;
7. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
8. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
10. import java.util.concurrent.TimeUnit;
12. public class AsyncTestJob {
13.     public static void main(String[] args) throws Exception {
15.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
17.         SingleOutputStreamOperator<JSONObject> stream = env.socketTextStream("localhost", 8888)
18.                 .map((MapFunction<String, JSONObject>) JSONObject::parseObject);
19.         
20.         AsyncDataStream.orderedWait(stream,
21.                 //使用异步算子
22.                 new MyAsyncFunction(50),
23.                 //超时时间
24.                 30000,
25.                 //超时时间单位
26.                 TimeUnit.MILLISECONDS,
27.                 //capacity为异步请求数参数。建议:该参数应小于等于线程池、连接池的最大连接数
28.                 50)
29.                 .print();
31.         env.execute();
32.     }
33. }

复制代码

1. package functions;
3. import com.alibaba.fastjson.JSONObject;
4. import com.zaxxer.hikari.HikariConfig;
5. import com.zaxxer.hikari.HikariDataSource;
6. import org.apache.flink.configuration.Configuration;
7. import org.apache.flink.streaming.api.functions.async.ResultFuture;
8. import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
9. import org.slf4j.Logger;
10. import org.slf4j.LoggerFactory;
12. import java.sql.Connection;
13. import java.sql.PreparedStatement;
14. import java.sql.SQLException;
15. import java.util.Collections;
16. import java.util.Properties;
17. import java.util.concurrent.*;
19. public class MyAsyncFunction extends RichAsyncFunction<JSONObject, String> {
20.     private final int maxConnTotal;
21.     private transient ExecutorService executorService;
22.     private HikariDataSource phoenixDataSource;
23.     private final static Logger logger = LoggerFactory.getLogger(AsyncUpsertHBase.class);
25.     public MyAsyncFunction(Integer maxConnTotal) {
26.         this.maxConnTotal = maxConnTotal;
27.     }
29.     @Override
30.     public void open(Configuration parameters) throws Exception {
31.         executorService = Executors.newFixedThreadPool(maxConnTotal);
32.         phoenixDataSource = getPhoenixDataSource();
33.     }
35.     @Override
36.     public void close() throws Exception {
37.         executorService.shutdown();
38.         phoenixDataSource.close();
39.     }
41.     @Override
42.     public void asyncInvoke(JSONObject json, ResultFuture<String> resultFuture) {
43.         //使用线程池实现异步提交请求
44.         Future<String> future = executorService.submit(() -> upsertHBase(json));
45.         CompletableFuture.supplyAsync(() -> {
46.             try {
47.                 return future.get();
48.             } catch (InterruptedException | ExecutionException e) {
49.                 return null;
50.             }
51.         }).thenAccept((String dbResult) -> resultFuture.complete(Collections.singleton(dbResult)));
52.     }
54.     /**
55.      * 标签写入HBase
56.      *
57.      * @param json 标签数据
58.      * @return 标签数据字符串
59.      * @throws SQLException SQLException
60.      */
61.     private String upsertHBase(JSONObject json) throws SQLException {
62.         Long id = json.getLong("id");
63.         String key = json.getString("key");
64.         Integer value = json.getInteger("data");
65.         String SQL = "UPSERT INTO TEST_TABLE (ID," + key + ") VALUES (" + id + "," + value + ")";
66.         try (
67.                 Connection conn = phoenixDataSource.getConnection();
68.                 PreparedStatement ps = conn.prepareStatement(SQL)
69.         ) {
70.             ps.executeUpdate();
71.             conn.commit();
72.             return json.toJSONString();
73.         } catch (Exception e) {
74.             logger.error(e.getMessage(), e);
75.             throw e;
76.         }
77.     }
79.     private HikariDataSource getPhoenixDataSource() {
80.         HikariConfig config = new HikariConfig();
81.         config.setDriverClassName("org.apache.phoenix.jdbc.PhoenixDriver");
82.         config.setJdbcUrl("jdbc:phoenix:10.xxx.xxx.xxx:2181");
83.         // 最大活跃连接数
84.         config.setMaximumPoolSize(50);
85.         //最小空闲连接数
86.         config.setMinimumIdle(10);
87.         //连接超时时间，单位毫秒
88.         config.setConnectionTimeout(30000);
89.         //空闲连接最大存活时间，单位毫秒
90.         config.setIdleTimeout(5 * 60 * 1000);
91.         //初始化失败时的超时时间，单位秒
92.         config.setInitializationFailTimeout(1);
93.         //池名称
94.         config.setPoolName("PhoenixHikariCP");
95.         // 允许池在达到最大大小时增长，如果当前所有连接都在使用中，并且执行者正在等待一个连接
96.         config.setAllowPoolSuspension(true);
97.         Properties properties = new Properties();
98.         properties.setProperty("phoenix.schema.mapSystemTablesToNamespace", "true");
99.         properties.setProperty("phoenix.schema.isNamespaceMappingEnabled", "true");
100.         properties.setProperty("phoenix.query.timeoutMs", "1200000");
101.         properties.setProperty("hbase.rpc.timeout", "1200000");
102.         properties.setProperty("hbase.client.scanner.timeout.period", "1200000");
103.         config.setDataSourceProperties(properties);
104.         return new HikariDataSource(config);
105.     }
106. }

复制代码

官方文档

地址：异步 I/O | Apache Flink
   超时处理 #

  当异步 I/O 哀求超时的时候，默认会抛出异常并重启作业。 假如你想处理超时，可以重写 AsyncFunction#timeout 方法。
  结果的顺序 #

  AsyncFunction 发出的并发哀求经常以不确定的顺序完成，这取决于哀求得到响应的顺序。 Flink 提供两种模式控制结果纪录以何种顺序发出。
  

• 无序模式： 异步哀求一竣事就立刻发出结果纪录。 流中纪录的顺序在颠末异步 I/O 算子之后发生了改变。 当利用 处理时间 作为基本时间特性时，这个模式具有最低的耽误和最少的开销。 此模式利用 AsyncDataStream.unorderedWait(...) 方法。
  
• 有序模式: 这种模式保持了流的顺序。发出结果纪录的顺序与触发异步哀求的顺序（纪录输入算子的顺序）相同。为了实现这一点，算子将缓冲一个结果纪录直到这条纪录前面的全部纪录都发出（或超时）。由于纪录大概结果要在 checkpoint 的状态中保存更长的时间，所以与无序模式相比，有序模式通常会带来一些额外的耽误和 checkpoint 开销。此模式利用 AsyncDataStream.orderedWait(...) 方法。
  

  事件时间 #

  当流处理应用利用事件时间时，异步 I/O 算子会精确处理 watermark。对于两种顺序模式，这意味着以下内容：
  

• 无序模式： Watermark 既不超前于纪录也不落后于纪录，即 watermark 建立了顺序的边界。 只有连续两个 watermark 之间的纪录是无序发出的。 在一个 watermark 背面天生的纪录只会在这个 watermark 发出以后才发出。 在一个 watermark 之前的全部输入的结果纪录全部发出以后，才会发出这个 watermark。
  这意味着存在 watermark 的情况下，无序模式 会引入一些与有序模式 相同的耽误和管理开销。开销巨细取决于 watermark 的频率。
  
• 有序模式： 连续两个 watermark 之间的纪录顺序也被保留了。开销与利用处理时间 相比，没有显著的差异。
  

  请记着，摄入时间 是一种特殊的事件时间，它基于数据源的处理时间自动天生 watermark。
  容错保证 #

  异步 I/O 算子提供了完全的精确一次容错保证。它将在途的异步哀求的纪录保存在 checkpoint 中，在故障恢复时重新触发哀求。
  实现提示 #

  在实现利用 Executor（大概 Scala 中的 ExecutionContext）和回调的 Futures 时，建议利用 DirectExecutor，因为通常回调的工作量很小，DirectExecutor 克制了额外的线程切换开销。回调通常只是把结果发送给 ResultFuture，也就是把它添加进输出缓冲。从这里开始，包括发送纪录和与 chenkpoint 交互在内的繁重逻辑都将在专有的线程池中进行处理。
  DirectExecutor 可以通过 org.apache.flink.util.concurrent.Executors.directExecutor() 或 com.google.common.util.concurrent.MoreExecutors.directExecutor() 获得。
  警告 #

  Flink 不以多线程方式调用 AsyncFunction
  我们想在这里明白指出一个经常肴杂的地方：AsyncFunction 不是以多线程方式调用的。 只有一个 AsyncFunction 实例，它被流中相应分区内的每个纪录顺序地调用。除非 asyncInvoke(...) 方法快速返回并且依靠于（客户端的）回调, 否则无法实现精确的异步 I/O。
  比方，以下情况导致壅闭的 asyncInvoke(...) 函数，从而使异步举动无效：
  

• 利用同步数据库客户端，它的查询方法调用在返回结果前一直被壅闭。
  
• 在 asyncInvoke(...) 方法内壅闭等待异步客户端返回的 future 范例对象
  

  目前，出于一致性的原因，AsyncFunction 的算子（异步等待算子）必须位于算子链的头部
  根据 FLINK-13063 给出的原因，目前我们必须断开异步等待算子的算子链以防止潜伏的一致性问题。这改变了先前支持的算子链的举动。必要旧有举动并继承可能违背一致性保证的用户可以实例化并手工将异步等待算子添加到作业图中并将链策略设置回通过异步等待算子的 ChainingStrategy.ALWAYS 方法进行链接。

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。