Realtime Compute for Apache Flink：Hudi

使用限制

• 僅Flink計算引擎vvr-4.0.11-flink-1.13及以上版本支援Hudi Connector。

• 檔案系統僅支援HDFS或阿里雲OSS和OSS-HDFS服務。

• 不支援以Session模式提交作業。

• 不支援修改欄位，如需修改，請在DLF控制台通過Spark SQL語句進行操作。

文法結構

CREATE TEMPORARY TABLE hudi_tbl (
  uuid BIGINT,
  data STRING,
  ts   TIMESTAMP(3),
  PRIMARY KEY(uuid) NOT ENFORCED
) WITH (
  'connector' = 'hudi',
  'path' = 'oss://<yourOSSBucket>/<自訂儲存位置>',
  ...
);

程式碼範例

• 源表

CREATE TEMPORARY TABLE blackhole (
  id INT NOT NULL PRIMARY KEY NOT ENFORCED,
  data STRING,
  ts TIMESTAMP(3)
) WITH (
  'connector' = 'blackhole'      
);

CREATE TEMPORARY TABLE hudi_tbl (
  id INT NOT NULL PRIMARY KEY NOT ENFORCED,
  data STRING,
  ts TIMESTAMP(3)
) WITH (
  'connector' = 'hudi', 
  'oss.endpoint' = '<yourOSSEndpoint>', 
  'accessKeyId' = '${secret_values.ak_id}', 
  'accessKeySecret' = '${secret_values.ak_secret}', 
  'path' = 'oss://<yourOSSBucket>/<自訂儲存位置>',
  'table.type' = 'MERGE_ON_READ',
  'read.streaming.enabled' = 'true'
);

-- 從最新的commit流讀寫入blackhole。
INSERT INTO blackhole SELECT * from hudi_tbl;

• 結果表

CREATE TEMPORARY TABLE datagen(
  id INT NOT NULL PRIMARY KEY NOT ENFORCED,
  data  STRING,
  ts TIMESTAMP(3)
) WITH (
  'connector' = 'datagen' ,
  'rows-per-second'='100' 
);

CREATE TEMPORARY TABLE hudi_tbl (
  id INT NOT NULL PRIMARY KEY NOT ENFORCED,
  data STRING,
  ts TIMESTAMP(3)
) WITH (
  'connector' = 'hudi', 
  'oss.endpoint' = '<yourOSSEndpoint>', 
  'accessKeyId' = '${secret_values.ak_id}', 
  'accessKeySecret' = '${secret_values.ak_secret}', 
  'path' = 'oss://<yourOSSBucket>/<自訂儲存位置>',
  'table.type' = 'MERGE_ON_READ'
);

INSERT INTO hudi_tbl SELECT * from datagen;

Datastream API

• maven pom

  根據使用的VVR版本，指定Flink和Hudi版本。

  <properties>
    <maven.compiler.source>8</maven.compiler.source>
    <maven.compiler.target>8</maven.compiler.target>
    <flink.version>1.15.4</flink.version>
    <hudi.version>0.13.1</hudi.version>
  </properties>
  
  <dependencies>
    <!-- flink -->
    <dependency>
      <groupId>org.apache.flink</groupId>
      <artifactId>flink-streaming-java</artifactId>
      <version>${flink.version}</version>
      <scope>provided</scope>
    </dependency>
    <dependency>
      <groupId>org.apache.flink</groupId>
      <artifactId>flink-table-common</artifactId>
      <version>${flink.version}</version>
      <scope>provided</scope>
    </dependency>
    <dependency>
      <groupId>org.apache.flink</groupId>
      <artifactId>flink-table-api-java-bridge</artifactId>
      <version>${flink.version}</version>
      <scope>provided</scope>
    </dependency>
    <dependency>
      <groupId>org.apache.flink</groupId>
      <artifactId>flink-table-planner_2.12</artifactId>
      <version>${flink.version}</version>
      <scope>provided</scope>
    </dependency>
    <dependency>
      <groupId>org.apache.flink</groupId>
      <artifactId>flink-clients</artifactId>
      <version>${flink.version}</version>
      <scope>provided</scope>
    </dependency>
  
    <!-- hudi -->
    <dependency>
      <groupId>org.apache.hudi</groupId>
      <artifactId>hudi-flink1.15-bundle</artifactId>
      <version>${hudi.version}</version>
      <scope>provided</scope>
    </dependency>
  
    <!-- oss -->
    <dependency>
      <groupId>org.apache.hadoop</groupId>
      <artifactId>hadoop-common</artifactId>
      <version>3.3.2</version>
      <scope>provided</scope>
    </dependency>
    <dependency>
      <groupId>org.apache.hadoop</groupId>
      <artifactId>hadoop-aliyun</artifactId>
      <version>3.3.2</version>
      <scope>provided</scope>
    </dependency>
  
    <!-- dlf -->
    <dependency>
      <groupId>com.aliyun.datalake</groupId>
      <artifactId>metastore-client-hive2</artifactId>
      <version>0.2.14</version>
      <scope>provided</scope>
    </dependency>
    <dependency>
      <groupId>org.apache.hadoop</groupId>
      <artifactId>hadoop-mapreduce-client-core</artifactId>
      <version>2.5.1</version>
      <scope>provided</scope>
    </dependency>
  </dependencies>

  重要

  DLF使用的部分依賴與社區版本存在衝突，例如hive-common、hive-exec。如果您有本地測試DLF的需求，可以下載hive-common和hive-execJAR包，然後在IDEA手動匯入。

• 寫入到Hudi

  import org.apache.flink.streaming.api.datastream.DataStream;
  import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  import org.apache.flink.table.data.GenericRowData;
  import org.apache.flink.table.data.RowData;
  import org.apache.flink.table.data.StringData;
  import org.apache.hudi.common.model.HoodieTableType;
  import org.apache.hudi.configuration.FlinkOptions;
  import org.apache.hudi.util.HoodiePipeline;
  
  import java.util.HashMap;
  import java.util.Map;
  
  public class FlinkHudiQuickStart {
  
    public static void main(String[] args) throws Exception {
      StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  
      String dbName = "test_db";
      String tableName = "test_tbl";
      String basePath = "oss://xxx";
  
      Map<String, String> options = new HashMap<>();
      // hudi conf
      options.put(FlinkOptions.PATH.key(), basePath);
      options.put(FlinkOptions.TABLE_TYPE.key(), HoodieTableType.MERGE_ON_READ.name());
      options.put(FlinkOptions.PRECOMBINE_FIELD.key(), "ts");
      options.put(FlinkOptions.DATABASE_NAME.key(), dbName);
      options.put(FlinkOptions.TABLE_NAME.key(), tableName);
      // oss conf
      options.put("hadoop.fs.oss.accessKeyId", "xxx");
      options.put("hadoop.fs.oss.accessKeySecret", "xxx");
      // 本地調試使用公網網端，例如oss-cn-hangzhou.aliyuncs.com；提交叢集使用內網網端，例如oss-cn-hangzhou-internal.aliyuncs.com
      options.put("hadoop.fs.oss.endpoint", "xxx");
      options.put("hadoop.fs.AbstractFileSystem.oss.impl", "org.apache.hadoop.fs.aliyun.oss.OSS");
      options.put("hadoop.fs.oss.impl", "org.apache.hadoop.fs.aliyun.oss.AliyunOSSFileSystem");
      // dlf conf
      options.put(FlinkOptions.HIVE_SYNC_ENABLED.key(), "true"); // 可選擇是否同步DLF
      options.put(FlinkOptions.HIVE_SYNC_MODE.key(), "hms");
      options.put(FlinkOptions.HIVE_SYNC_DB.key(), dbName);
      options.put(FlinkOptions.HIVE_SYNC_TABLE.key(), tableName);
      options.put("hadoop.dlf.catalog.id", "xxx");
      options.put("hadoop.dlf.catalog.accessKeyId", "xxx");
      options.put("hadoop.dlf.catalog.accessKeySecret", "xxx");
      options.put("hadoop.dlf.catalog.region", "xxx");
      //  本地調試使用公網網端，例如dlf.cn-hangzhou.aliyuncs.com，提交叢集使用內網網端，例如dlf-vpc.cn-hangzhou.aliyuncs.com
      options.put("hadoop.dlf.catalog.endpoint", "xxx");
      options.put("hadoop.hive.imetastoreclient.factory.class", "com.aliyun.datalake.metastore.hive2.DlfMetaStoreClientFactory");
  
      DataStream<RowData> dataStream = env.fromElements(
          GenericRowData.of(StringData.fromString("id1"), StringData.fromString("name1"), 22,
              StringData.fromString("1001"), StringData.fromString("p1")),
          GenericRowData.of(StringData.fromString("id2"), StringData.fromString("name2"), 32,
              StringData.fromString("1002"), StringData.fromString("p2"))
      );
  
      HoodiePipeline.Builder builder = HoodiePipeline.builder(tableName)
          .column("uuid string")
          .column("name string")
          .column("age int")
          .column("ts string")
          .column("`partition` string")
          .pk("uuid")
          .partition("partition")
          .options(options);
  
      builder.sink(dataStream, false); // The second parameter indicating whether the input data stream is bounded
      env.execute("Flink_Hudi_Quick_Start");
    }
  }

常見問題

• 儲存中一直查不到資料，應該如何處理？

• 資料有重複時，應該如何去做？

• 為什麼Merge On Read只有log檔案？