Flink SQL中使用异步io关联维表

一、目标

希望使用flink sql来关联维度表，但是想用异步IO的方式关联。

二、方案

当前社区进展：目前Flink SQL 中的connector都没实现异步io关联维表，但是接口是已经支持了的，可以自定义实现；HBase connector 社区有人正在支持异步io关联维表，预计1.13可以使用。

2.1、自己实现demo：

    /*** An async lookup function which find matched rows with the given fields. NOTE: We have to* declare it as public because it will be used in code generation.*/public static class AsyncTestValueLookupFunction extends AsyncTableFunction<Row> {private static final long serialVersionUID = 1L;private final Map<Row, List<Row>> mapping;private transient boolean isOpenCalled = false;private transient ExecutorService executor;protected AsyncTestValueLookupFunction(Map<Row, List<Row>> mapping) {this.mapping = mapping;}@Overridepublic void open(FunctionContext context) throws Exception {RESOURCE_COUNTER.incrementAndGet();isOpenCalled = true;executor = Executors.newSingleThreadExecutor();}public void eval(CompletableFuture<Collection<Row>> resultFuture, Object... inputs) {checkArgument(isOpenCalled, "open() is not called.");final Row key = Row.of(inputs);if (Arrays.asList(inputs).contains(null)) {throw new IllegalArgumentException(String.format("Lookup key %s contains null value, which should not happen.",key));}CompletableFuture.supplyAsync(() -> {List<Row> list = mapping.get(key);if (list == null) {return Collections.<Row>emptyList();} else {return list;}},executor).thenAccept(resultFuture::complete);}@Overridepublic void close() throws Exception {RESOURCE_COUNTER.decrementAndGet();if (executor != null && !executor.isShutdown()) {executor.shutdown();}}}

2.2、社区实现：

https://github.com/apache/flink/pull/14684#pullrequestreview-604148209

Flink SQL中使用异步io关联维表相关推荐

Flink（54）：Flink高级特性之异步IO（Async I/O）
目录 0. 相关文章链接 1. 异步IO概述 1.1. 异步IO操作的需求 1.2. 使用Aysnc I/O的前提条件 1.3. Async I/O API 2. 案例展示 2.1. 需求 2.2. ...
linux 内核io操作,关于Linux内核中的异步IO的使用
我们都知道异步IO的作用,就是可以提高我们程序的并发能力,尤其在网络模型中.在linux中有aio的一系列异步IO的函数接口,但是这类函数都是glibc库中的函数,是基于多线程实现,不是真正的异步IO ...
python 异步io_Python中的异步IO：完整的演练
python 异步io Async IO is a concurrent programming design that has received dedicated support in Pytho ...
【Flink】Flink 源码阅读笔记（18）- Flink SQL 中的流和动态表
1.概述转载:Flink 源码阅读笔记(18)- Flink SQL 中的流和动态表
Flink SQL中的函数
Table API是内嵌在Java语言中的,很多方法需要在类中额外添加,扩展功能比较麻烦,目前支持的函数比较少,故一般情况下我们使用Flink SQL中的函数 Flink SQL中的函数主要分为两类: ...
flink sql 如何upsert 到一张hologres表中
Flink Table 的三种 Sink 模式作为计算引擎 Flink 应用的计算结果总要以某种方式输出,比如调试阶段的打印到控制台或者生产阶段的写到数据库.而对于本来就需要在 Flink 内存保存 ...
KNIME中使用Unpivoting将二维表转为一维表
出现问题因为是中文数据,使用CSV READER读取后出现乱码,同时Unpivoting中也出现乱码. 在该NOTE中设置字符: 原始数据原表数据,从图中可以看出是二维表,不利于数据分析,目标是保 ...
Fink异步IO的实战(关联维表)
简介异步io实战知识前提线程池异步io 应用程序 public class ASyncIODemo {public static void main(String[] args) throws ...
flink批处理访问mysql_Flink 异步IO访问外部数据（mysql篇）
最近看了大佬的博客,突然想起Async I/O方式是Blink 推给社区的一大重要功能,可以使用异步的方式获取外部数据,想着自己实现以下,项目上用的时候,可以不用现去找了. 最开始想用scala 实现 ...

Flink SQL中使用异步io关联维表

目录

一、目标

二、方案

2.1、自己实现demo：

2.2、社区实现：

Flink SQL中使用异步io关联维表相关推荐

最新文章

热门文章

Flink SQL中使用异步io关联维表

目录

一、目标

二、方案

2.1、 自己实现demo：

2.2、 社区实现：

Flink SQL中使用异步io关联维表相关推荐

最新文章

热门文章

2.1、自己实现demo：

2.2、社区实现：