Flink Hudi分析

本代码分析基于Flink1.13.1、Hudi0.10.0

-- 生成数据
create table datagen (
    id           bigint,
    prod_id      bigint,
    price        int,
    uid          bigint,
    buy_time     timestamp
) with (
  'connector' = 'datagen'
);
-- hudi数据表
create table t_hudi (
    id           bigint,
    prod_id      bigint,
    price        int,
    uid          bigint,
    buy_time     timestamp
) with (
  'connector' = 'hudi',
  'path' = '${path}',
  'table.type' = 'MERGE_ON_READ'
);

insert into t_hudi select * from datagen;

通过分析Hudi connector，meta-INF/services/org.apache.flink.table.factories.Factory中

（HoodieTableFactory、HoodieTableSink的内容，涉及 Flink dynamic table)

发现Flink Hudi写入时所有流程都在HoodieTableSink中

下面是对HoodieTableSink Pipelines的分析：涉及bootstrap、hoodieStreamWrite、compact

Flink流处理调用streamBootstrap

  public static DataStream bootstrap(
      Configuration conf,
      RowType rowType,
      int defaultParallelism,
      DataStream dataStream,
      boolean bounded,
      boolean overwrite) {
    final boolean globalIndex = conf.getBoolean(FlinkOptions.INDEX_GLOBAL_ENABLED);
    if (overwrite) {
      return rowDataToHoodieRecord(conf, rowType, dataStream);
    } else if (bounded && !globalIndex && OptionsResolver.isPartitionedTable(conf)) {
      return boundedBootstrap(conf, rowType, defaultParallelism, dataStream);
    } else {
      return streamBootstrap(conf, rowType, defaultParallelism, dataStream, bounded);
    }
  }

  private static DataStream streamBootstrap(
      Configuration conf,
      RowType rowType,
      int defaultParallelism,
      DataStream dataStream,
      boolean bounded) {
    DataStream dataStream1 = rowDataToHoodieRecord(conf, rowType, dataStream);

    if (conf.getBoolean(FlinkOptions.INDEX_BOOTSTRAP_ENABLED) || bounded) {
      dataStream1 = dataStream1
          .transform(
              "index_bootstrap",
              TypeInformation.of(HoodieRecord.class),
              new BootstrapOperator<>(conf))
          .setParallelism(conf.getOptional(FlinkOptions.INDEX_BOOTSTRAP_TASKS).orElse(defaultParallelism))
          .uid("uid_index_bootstrap_" + conf.getString(FlinkOptions.TABLE_NAME));
    }

    return dataStream1;
  }

  public static DataStream rowDataToHoodieRecord(Configuration conf, RowType rowType, DataStream dataStream) {
    return dataStream.map(RowDataToHoodieFunctions.create(rowType, conf), TypeInformation.of(HoodieRecord.class));
  }

  private HoodieRecord toHoodieRecord(I record) throws Exception {
    GenericRecord gr = (GenericRecord) this.converter.convert(this.avroSchema, record);
    final HoodieKey hoodieKey = keyGenerator.getKey(gr);

    HoodieRecordPayload payload = payloadCreation.createPayload(gr);
    HoodieOperation operation = HoodieOperation.fromValue(record.getRowKind().toBytevalue());
    return new HoodieRecord<>(hoodieKey, payload, operation);
  }

这里只展开讲解RowDataToHoodieFunction，它是负责将Flink RowData数据转化成HoodieRecord

1.首先将RowData转成GenericRecord（avro格式）

2.根据规则生成HoodieKey

3.根据RowData的RowKind获取HoodieOperation，是hudi可以处理增删改事件

1.加载hudi索引

2.waitForBootstrapReady等待其他subTask准备完成

  public static DataStream