实例

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
StreamExecutionEnvironment.createLocalEnvironment();
TableLoader tableLoader = TableLoader.fromHadoopTable("XXX");DataStream<RowData> stream = FlinkSource.forRowData().env(env).tableLoader(tableLoader).streaming(true).build();stream.print();
env.execute("IcebergRead");

流程详解

首先看build()方法：

public DataStream<RowData> build() {Preconditions.checkNotNull(this.env, "StreamExecutionEnvironment should not be null");FlinkInputFormat format = this.buildFormat();ScanContext context = this.contextBuilder.build();TypeInformation<RowData> typeInfo = FlinkCompatibilityUtil.toTypeInfo(FlinkSchemaUtil.convert(context.project()));if (!context.isStreaming()) {int parallelism = this.inferParallelism(format, context);if (this.env.getMaxParallelism() > 0) {parallelism = Math.min(parallelism, this.env.getMaxParallelism());}return this.env.createInput(format, typeInfo).setParallelism(parallelism);} else {StreamingMonitorFunction function = new StreamingMonitorFunction(this.tableLoader, context);String monitorFunctionName = String.format("Iceberg table (%s) monitor", this.table);String readerOperatorName = String.format("Iceberg table (%s) reader", this.table);return this.env.addSource(function, monitorFunctionName).transform(readerOperatorName, typeInfo, StreamingReaderOperator.factory(format));}
}

该方法主要做了两件事情：

buildFormat()方法，利用tableLoader加载对应的table，然后通过这个table获取到对应的Schema、FileIO、EncryptionManager；再加上contextBuilder.build()方法构建出的ScanContext对象，一起组装成了负责辅助DataSource读取数据、分发数据的InputFormat；然后分两种情况，批读取和流读取
在流式读取情况下，将StreamingMonitorFunction和StreamingReaderOperator算子注册到env上

StreamingMonitorFunction：不停的扫描iceberg表看是否有新的snapshot生成，如果有则生成CombinedScanTask发向下游。
StreamingReaderOperator：一旦收到source发来的split，会将其放到一个队列中，然后通过一个MailboxExecutor线程处理，这种结构可以将读取数据和处理checkpoint barriers功能分离，避免潜在的背压。

StreamingMonitorFunction

继承关系：

它实现了CheckpointedFunction接口，所以能够保证在source端的一致性；
另外，因为它并没有实现ParallelSourceFunction接口，所以它注定只能有一个并行度。这里的目的是确保在只有一个线程去监控Iceberg表和分发任务，多线程只会发生数据错乱。
run()方法流程：

public void run(SourceFunction.SourceContext<FlinkInputSplit> ctx) throws Exception {this.sourceContext = ctx;while(this.isRunning) {this.monitorAndForwardSplits();Thread.sleep(this.scanContext.monitorInterval().toMillis());}}

monitorAndForwardSplits()方法，获取表当前最新的快照snapshotId，如果记录了lastSnapshotId，那就生成lastSnapshotId到snapshotId之间的增量文件的FlinkInputSplit对象：

void monitorAndForwardSplits() {this.table.refresh();Snapshot snapshot = this.table.currentSnapshot();if (snapshot != null && snapshot.snapshotId() != this.lastSnapshotId) {long snapshotId = snapshot.snapshotId();ScanContext newScanContext;if (this.lastSnapshotId == -1L) {newScanContext = this.scanContext.copyWithSnapshotId(snapshotId);} else {snapshotId = this.toSnapshotIdInclusive(this.lastSnapshotId, snapshotId, this.scanContext.maxPlanningSnapshotCount());newScanContext = this.scanContext.copyWithAppendsBetween(this.lastSnapshotId, snapshotId);}LOG.debug("Start discovering splits from {} (exclusive) to {} (inclusive)", this.lastSnapshotId, snapshotId);long start = System.currentTimeMillis();FlinkInputSplit[] splits = FlinkSplitPlanner.planInputSplits(this.table, newScanContext, this.workerPool);LOG.debug("Discovered {} splits, time elapsed {}ms", splits.length, System.currentTimeMillis() - start);start = System.currentTimeMillis();synchronized(this.sourceContext.getCheckpointLock()) {FlinkInputSplit[] var9 = splits;int var10 = splits.length;int var11 = 0;while(true) {if (var11 >= var10) {this.lastSnapshotId = snapshotId;break;}FlinkInputSplit split = var9[var11];this.sourceContext.collect(split);++var11;}}LOG.debug("Forwarded {} splits, time elapsed {}ms", splits.length, System.currentTimeMillis() - start);}}

此处核心：

构造出从startSnapshotId到snapshotId之间的增量FlinkInputSplit（FlinkSplitPlanner.planInputSplits为核心内容，流程二详细讲解）
将FlinkInputSplit分配给下游进一步的处理

StreamingReaderOperator

继承关系：

一些参数：

private final MailboxExecutor executor;
private FlinkInputFormat format;
private transient SourceFunction.SourceContext<RowData> sourceContext;
private transient ListState<FlinkInputSplit> inputSplitsState;
private transient Queue<FlinkInputSplit> splits;
private transient SplitState currentSplitState;

其中：

executor是暴露出来的一个执行器，这个线程同时处理用户操作和checkpoint动作，我们一次只预定一个InputSplit去读取，因此当新的checkpoint到达是能被触发而不是被InputSplit读取操作阻塞。
inputSplitsState为存储FlinkInputSplit的状态变量，即需要被读取的FlinkInputSplit，会在checkpoint持久化。
splits为当前周期需要读取的FlinkInputSplit，会在initializeState从inputSplitsState读出来。
currentSplitState表示当前的读取状态。

处理数据流程：

public void processElement(StreamRecord<FlinkInputSplit> element) {this.splits.add((FlinkInputSplit)element.getValue());this.enqueueProcessSplits();
}

将接收到的数据加入splits然后调用enqueueProcessSplits方法

private void enqueueProcessSplits() {if (this.currentSplitState == StreamingReaderOperator.SplitState.IDLE && !this.splits.isEmpty()) {this.currentSplitState = StreamingReaderOperator.SplitState.RUNNING;this.executor.execute(this::processSplits, this.getClass().getSimpleName());}}

在executor中异步的执行了如下操作：

从列表头中取出一个FlinkInputSplit对象，调用FlinkInputFormat.open()
轮询调用FlinkInputFormat.nextRecord()获取RowData数据对象，并交给了flink的SourceContext，至此数据真正的进入了流
一直循环1-2这个过程，直到队列为空。

private void processSplits() throws IOException {FlinkInputSplit split = (FlinkInputSplit)this.splits.poll();if (split == null) {this.currentSplitState = StreamingReaderOperator.SplitState.IDLE;} else {this.format.open(split);try {RowData nextElement = null;while(!this.format.reachedEnd()) {nextElement = this.format.nextRecord(nextElement);this.sourceContext.collect(nextElement);}} finally {this.currentSplitState = StreamingReaderOperator.SplitState.IDLE;this.format.close();}this.enqueueProcessSplits();}
}

StreamingReaderOperator中有一个成员变量为FlinkInputFormat format，FlinkInputFormat继承自flink中的RichInputFormat，RichInputFormat继承自InputFormat，InputFormat为读取数据时候的一个抽象类，一些数据的读取数据的相关类都基于它实现。
format的open()方法会去构建一个DataIterator对象，DataIterator对应一个CombinedScanTask的数据读取的迭代器：

public void open(FlinkInputSplit split) {this.iterator = new DataIterator(this.rowDataReader, split.getTask(), this.io, this.encryption);
}

nextRecord()方法获取下一个元素：

public RowData nextRecord(RowData reuse) {++this.currentReadCount;return (RowData)this.iterator.next();
}

进入DataIterator.next()：

public T next() {this.updateCurrentIterator();++this.recordOffset;return this.currentIterator.next();
}private void updateCurrentIterator() {try {while(!this.currentIterator.hasNext() && this.tasks.hasNext()) {this.currentIterator.close();this.currentIterator = this.openTaskIterator((FileScanTask)this.tasks.next());++this.fileOffset;this.recordOffset = 0L;}} catch (IOException var2) {throw new UncheckedIOException(var2);}
}private CloseableIterator<T> openTaskIterator(FileScanTask scanTask) {return this.fileScanTaskReader.open(scanTask, this.inputFilesDecryptor);
}

updateCurrentIterator()函数轮询了CombinedScanTask中的Collection files()，针对每个FileScanTask执行了FileScanTaskReader的fileScanTaskReader.open(scanTask, inputFilesDecryptor)，通过FileScanTask任务读取了RowData对象，读取底层文件，包括PARQUET、AVRO、ORC三种文件格式的读取。

Iceberg源码学习：flink读iceberg流程一相关推荐

读文章笔记（三）：从源码学习Transformer
读文章笔记(三):从源码学习Transformer encoder分为两部分: decoder 公众号机器学习算法工程师文章链接: https://mp.weixin.qq.com/s/0NajB_ ...
第三课 k8s源码学习和二次开发-缓存机制Informers和Reflector组件学习
第三课 k8s源码学习和二次开发-缓存机制Informers和Reflector组件学习 tags: k8s 源码学习 categories: 源码学习二次开发文章目录第三课 k8s源码学习和二 ...
Shiro源码学习之一
一.最基本的使用 1.Maven依赖 <dependency><groupId>org.apache.shiro</groupId><artifactId&g ...
Vuex源码学习（五）加工后的module
没有看过moduleCollection那可不行!Vuex源码学习(四)module与moduleCollection 感谢提出代码块和截图建议的小伙伴代码块和截图的区别: 代码块部分希望大家按照我 ...
Nginx源码研究之nginx限流模块详解
这篇文章主要介绍了Nginx源码研究之nginx限流模块详解,小编觉得挺不错的,现在分享给大家,也给大家做个参考.一起跟随小编过来看看吧高并发系统有三把利器:缓存.降级和限流: 限流的目的是通过对并 ...
ConcurrentHashMap源码学习
ConcurrentHashMap源码学习自从学习了AQS之后,想着重新读一下ConcurrentHashMap的源码来加深下理解,所以有了这篇文章,针对ConcurrentHashMap常用的方法 ...
ASP.NET Core MVC 源码学习：MVC 启动流程详解
前言在上一篇文章中,我们学习了 ASP.NET Core MVC 的路由模块,那么在本篇文章中,主要是对 ASP.NET Core MVC 启动流程的一个学习. ASP.NET Core 是新一 ...
MVC系列——MVC源码学习：打造自己的MVC框架（一：核心原理）(转)
阅读目录一.MVC原理解析 1.MVC原理二.HttpHandler 1.HttpHandler.IHttpHandler.MvcHandler的说明 2.IHttpHandler解析 3.Mvc ...
Promise源码学习（2）
Promise源码学习(2) 本篇为上一篇源码学习(1)的补充,主要是来介绍Promise.all()和Promise.race()方法. 闲话少叙,进入正题 Promise.race() 首先来简单 ...

Iceberg源码学习：flink读iceberg流程一

实例

流程详解

StreamingMonitorFunction

StreamingReaderOperator

Iceberg源码学习：flink读iceberg流程一相关推荐

最新文章

热门文章