江苏龙网

  1. 首页 > 人文 > >

Apache Hudi与Apache Flink集成( 二 )


这批操作最容易想到的是通过使用时间窗口来实现 ， 然而 ， 使用窗口 ， 在某个窗口没有数据流入时 ， 将没有输出数据 ， Sink端难以判断同一批数据是否已经处理完 。 因此我们使用flink的检查点机制来攒批 ， 每两个barrier之间的数据为一个批次 ， 当某个子任务中没有数据时 ， mock结果数据凑数 。 这样在Sink端 ， 当每个子任务都有结果数据下发时即可认为一批数据已经处理完成 ， 可以执行commit 。
DAG如下：
Apache Hudi与Apache Flink集成文章插图

  • source 接收kafka数据 ， 转换成List;
  • InstantGeneratorOperator 生成全局唯一的instant.当上一个instant未完成或者当前批次无数据时 ， 不创建新的instant；
  • KeyBy partitionPath 根据 partitionPath分区 ， 避免多个子任务写同一个分区；
  • WriteProcessOperator 执行写操作 ， 当当前分区无数据时 ， 向下游发送空的结果数据凑数；
  • CommitSink 接收上游任务的计算结果 ， 当收到 parallelism个结果时 ， 认为上游子任务全部执行完成 ， 执行commit.
注：InstantGeneratorOperator和WriteProcessOperator 均为自定义的Flink算子 ， InstantGeneratorOperator会在其内部阻塞检查上一个instant的状态 ， 保证全局只有一个inflight（或requested）状态的instant.WriteProcessOperator是实际执行写操作的地方 ， 其写操作在checkpoint时触发 。
5. 实现示例1) HoodieTable/** * Abstract implementation of a HoodieTable. * * @paramSub type of HoodieRecordPayload * @param Type of inputs * @param Type of keys * @param Type of outputs */public abstract class HoodieTable implements Serializable {protected final HoodieWriteConfig config;protected final HoodieTableMetaClient metaClient;protected final HoodieIndex index;public abstract HoodieWriteMetadata upsert(HoodieEngineContext context, String instantTime,I records);public abstract HoodieWriteMetadata insert(HoodieEngineContext context, String instantTime,I records);public abstract HoodieWriteMetadata bulkInsert(HoodieEngineContext context, String instantTime,I records, Option> bulkInsertPartitioner);......}HoodieTable 是 hudi的核心抽象之一 ， 其中定义了表支持的insert,upsert,bulkInsert等操作 。 以 upsert 为例 ， 输入数据由原先的 JavaRDD inputRdds 换成了 I records, 运行时 JavaSparkContext jsc 换成了 HoodieEngineContext context.
从类注释可以看到 T,I,K,O分别代表了hudi操作的负载数据类型、输入数据类型、主键类型以及输出数据类型 。 这些泛型将贯穿整个抽象层 。
2) HoodieEngineContext/** * Base class contains the context information needed by the engine at runtime. It will be extended by different * engine implementation if needed. */public abstract class HoodieEngineContext {public abstract List map(List data, SerializableFunction func, int parallelism);public abstract List flatMap(List data, SerializableFunction> func, int parallelism);public abstract void foreach(List data, SerializableConsumer consumer, int parallelism);......}【Apache Hudi与Apache Flink集成】HoodieEngineContext 扮演了 JavaSparkContext 的角色 ， 它不仅能提供所有 JavaSparkContext能提供的信息 ， 还封装了 map,flatMap,foreach等诸多方法 ， 隐藏了JavaSparkContext#map(),JavaSparkContext#flatMap(),JavaSparkContext#foreach()等方法的具体实现 。


推荐阅读


上一篇：阿里巴巴要开火锅店了？注册盒马火锅，能像盒马鲜生一样成功吗？

下一篇：双模5G仅1716元，还有4800万三摄，可惜依旧卖不动