Realtime Compute for Apache Flink:Hudi コネクタ (廃止予定)

制限事項

• エンジンバージョンが vvr-4.0.11-flink-1.13 以降の Realtime Compute for Apache Flink のみが Hudi コネクタをサポートしています。

• HDFS、Alibaba Cloud OSS、または OSS-HDFS のみがファイルシステムとして使用できます。

• セッションクラスタではドラフトを公開できません。

• Hudi コネクタを使用してテーブルのフィールドを変更することはできません。テーブルのフィールドを変更する場合は、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>/<Custom storage directory>',
  'table.type' = 'MERGE_ON_READ',
  'read.streaming.enabled' = 'true'
);

-- ストリーミングモードで最新のコミットからデータを読み取り、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>/<Custom storage directory>',
  'table.type' = 'MERGE_ON_READ'
);

INSERT INTO hudi_tbl SELECT * from datagen;

DataStream API

• maven pom

  使用されている VVR バージョンに基づいて、Realtime Compute for Apache 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 などのオープンソースの Hive バージョンと競合します。オンプレミス環境で DLF をテストする場合は、hive-common および hive-exec JAR パッケージをダウンロードし、IntelliJ 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"); // Hudi データを 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); // 2 番目のパラメーターは、入力データストリームが有界かどうかを示します
      env.execute("Flink_Hudi_Quick_Start");
    }
  }

よくある質問

• ストレージにデータが見つからない場合はどうすればよいですか？

• 重複データが存在する場合はどうすればよいですか？

• MOR モードではログファイルのみが生成されるのはなぜですか？