生活_傻大方

  1. 首页 > 潮·科技 > >

Apache Hudi与Apache Flink集成( 二 )

  • 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);......}HoodieEngineContext 扮演了 JavaSparkContext 的角色 ， 它不仅能提供所有 JavaSparkContext能提供的信息 ， 还封装了 map,flatMap,foreach等诸多方法 ， 隐藏了JavaSparkContext#map(),JavaSparkContext#flatMap(),JavaSparkContext#foreach()等方法的具体实现 。
    以map方法为例 ， 在Spark的实现类 HoodieSparkEngineContext中 ， map方法如下：
    @Overridepublic List map(List data, SerializableFunction func, int parallelism) {return javaSparkContext.parallelize(data, parallelism).map(func::apply).collect();}在操作List的引擎中其实现可以为（不同方法需注意线程安全问题 ， 慎用parallel()）：
    @Overridepublic List map(List data, SerializableFunction func, int parallelism) {return data.stream().parallel().map(func::apply).collect(Collectors.toList());}注：map函数中抛出的异常 ， 可以通过包装SerializableFunction func解决.
    这里简要介绍下 SerializableFunction:
    @FunctionalInterfacepublic interface SerializableFunction extends Serializable {O apply(I v1) throws Exception;}该方法实际上是 java.util.function.Function 的变种 ， 与java.util.function.Function 不同的是 SerializableFunction可以序列化 ， 可以抛异常 。 引入该函数是因为JavaSparkContext#map()函数能接收的入参必须可序列 ， 同时在hudi的逻辑中 ， 有多处需要抛异常 ， 而在Lambda表达式中进行 try catch 代码会略显臃肿 ， 不太优雅 。
    6.现状和后续计划6.1 工作时间轴2020年4月 ， T3出行（杨华@vinoyang ， 王祥虎@wangxianghu）和阿里巴巴的同学（李少锋@leesf）以及若干其他小伙伴一起设计、敲定了该解耦方案；
    2020年4月 ， T3出行(王祥虎@wangxianghu)在内部完成了编码实现 ， 并进行了初步验证 ， 得出方案可行的结论；
    2020年7月 ， T3出行(王祥虎@wangxianghu)将该设计实现和基于新抽象实现的Spark版本推向社区（HUDI-1089）；
    2020年9月26日 ， 顺丰科技基于T3内部分支修改完善的版本在 Apache Flink Meetup（深圳站）公开PR, 使其成为业界第一个在线上使用Flink将数据写hudi的企业 。
    2020年10月2日 ， HUDI-1089 合并入hudi主分支 ， 标志着hudi-spark解耦完成 。
    6.2 后续计划1）推进hudi和flink集成
    将flink与hudi的集成尽快推向社区,在初期 ， 该特性可能只支持kafka数据源 。
    2）性能优化
    为保证hudi-spark版本的稳定性和性能 ， 此次解耦没有太多考虑flink版本可能存在的性能问题 。



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

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