flink流式增量查询hudi表流程分析

　　环境flink 1.13.6 hudi 0.11.0 merge on read 表
　　代码示例tEnv.executeSql(＂CREATE  TABLE tb_person_hudi ( id BIGINT, age INT, name STRING,create_time TIMESTAMP ( 3 ), time_stamp TIMESTAMP(3),PRIMARY KEY ( id ) NOT ENFORCED ) WITH ( ＂ +         ＂	＂connector＂ = ＂hudi＂, ＂ +         ＂	＂table.type＂ = ＂MERGE_ON_READ＂, ＂ +         ＂	＂path＂ = ＂file:///D:/data/hadoop3.2.1/warehouse/tb_person_hudi＂, ＂ +         ＂	＂read.start-commit＂ = ＂20220722103000＂, ＂ +       //  ＂	＂read.end-commit＂ = ＂20220722104000＂, ＂ +         ＂	＂read.task＂ = ＂1＂, ＂ +         ＂	＂read.streaming.enabled＂ = ＂true＂, ＂ +         ＂	＂read.streaming.check-interval＂ = ＂30＂  ＂ +         ＂)＂); Table table = tEnv.sqlQuery(＂select * from tb_person_hudi ＂); tEnv.toChangelogStream(table).print().setParallelism(1); env.execute(＂test＂);
　　流程分析
　　hudi源入口(HoodieTableSource)
　　HoodieTableSource实现ScanTableSource,SupportsPartitionPushDown,SupportsProjectionPushDown,SupportsLimitPushDown,SupportsFilterPushDown接口，后4个接口主要是支持对查询计划的优化。ScanTableSource则提供了读取hudi表的具体实现，核心方法为org.apache.hudi.table.HoodieTableSource#getScanRuntimeProvider ：if (conf.getBoolean(FlinkOptions.READ_AS_STREAMING)) {  //开启了流式读(read.streaming.enabled)   StreamReadMonitoringFunction monitoringFunction = new StreamReadMonitoringFunction(       conf, FilePathUtils.toFlinkPath(path), maxCompactionMemoryInBytes, getRequiredPartitionPaths());   InputFormat inputFormat = getInputFormat(true);   OneInputStreamOperatorFactory factory = StreamReadOperator.factory((MergeOnReadInputFormat) inputFormat);   SingleOutputStreamOperator source = execEnv.addSource(monitoringFunction, getSourceOperatorName(＂split_monitor＂))       .setParallelism(1)       .transform(＂split_reader＂, typeInfo, factory)       .setParallelism(conf.getInteger(FlinkOptions.READ_TASKS));    return new DataStreamSource<>(source); }
　　上面代码在流环境中创建了一个SourceFunction(StreamReadMonitoringFunction)和一个自定义的转换(StreamReadOperator)StreamReadMonitoringFunction: 监控hudi表元数据目录(.hoodie)获取需要被读取的文件分片(MergeOnReadInputSplit，一个base parquet文件和一组log文件)，然后把分片递给下游的转换算子StreamReadOperator进行文件读取；固定一个线程去监控，名称为split_monitorxxxxx. StreamReadOperator：将按timeline升序收到的MergeOnReadInputSplit一个一个地读取分片数据；算子名称为split_reader->xxxxx，可以通过设置read.tasks进行设置并行度
　　定时监控元数据获得增量分片(StreamReadMonitoringFunction)
　　StreamReadMonitoringFunction负责定时(read.streaming.check-interval)扫描hudi表的元数据目录.hoodie，如果发现在active timeline上有新增的instant[action=commit,deltacommit,compaction,replace && active=completed]，从这些instant信息中可以知道数据变更写到了哪些文件(parquet,log)，然后构建成分片对象(MergeOnReadInputSplit)。核心属性：issuedInstant，这个是增量查询的依据，记录着当前已经消费的数据的最新instant，类似于kafka的offset，但是hudi是基于timeline.该值是有状态的，维护在ListState中，所以flink job重启依然可以做到增量。 核心方法：StreamReadMonitoringFunction#monitorDirAndForwardSplits，很简单，就做了两件事，调用IncrementalInputSplits#inputSplits获取到增量分片(有序)，然后传递给下游的算子(StreamReadOperator) public void monitorDirAndForwardSplits(SourceContext context) {   HoodieTableMetaClient metaClient = getOrCreateMetaClient();   IncrementalInputSplits.Result result =       incrementalInputSplits.inputSplits(metaClient, this.hadoopConf, this.issuedInstant);    for (MergeOnReadInputSplit split : result.getInputSplits()) {     context.collect(split);   } }
　　获取增量分片(IncrementalInputSplits)
　　主要逻辑在方法IncrementalInputSplits#inputSplits(metaClient, hadoopConf, issuedInstant)，需要先了解hudi关于timeline和instant的一些基本概念，详细的流程如下图所示：
　　如果flink job首次运行指定了read.start-commit和read.end-commit，但是该范围是比较久以前，instant已经被归档，那么流作业将永远不能消费到数据
　　https://github.com/apache/hudi/issues/6167
　　读取数据文件(StreamReadOperator)
　　StreamReadOperator算子接收分片后会缓存在队列Queue splits，然后不停从队列中poll分片放到线程池中执行
　　private void processSplits() throws IOException { format.open(split); consumeAsMiniBatch(split); enqueueProcessSplits(); }
　　主要有三个步骤：从队列中peek分片，调用MergeOnReadInputFormat.open构建迭代器，迭代器是用来进行文件的数据读取，一个迭代器对应一个分片（多个物理文件，base+log），对应不同读取的场景，有几种迭代器：BaseFileOnlyFilteringIterator,BaseFileOnlyIterator,LogFileOnlyIterator,MergeIterator,SkipMergeIterator 微批量消费，每批只读2048记录，将把记录传递给下游的算子消费同时标记消费的总数，如果该分片读到了尾，则将该分片从队列中弹出，并关闭MergeOnReadInputFormat 继续处理队列中的分片，回到步骤1，如果上一次的分片没消费完，那么本次循环将继续消费，只不过是由另一个线程处理。