MaxCompute:Flink を使用した MaxCompute へのデータ書き込み

背景情報

• サポートされている書き込みモード

  Flink コネクタは、アップサートと挿入の 2 つの書き込みモードをサポートしています。アップサートモードでは、次のいずれかの方法でデータストリームをグループ化できます。

  • プライマリキーによるグループ化

  • パーティションフィールドによるグループ化

    パーティションフィールドによるグループ化は、多数のパーティションに適していますが、データスキューを引き起こす可能性があります。

• アップサートの書き込み手順と推奨パラメーターについては、「データウェアハウスへのリアルタイムデータインジェスト」をご参照ください。

• 書き込みモードは Flink コネクタのパラメーターを使用して指定します。コネクタパラメーターの完全なリストについては、「付録： Flink コネクタのパラメーター」をご参照ください。

• Flink のアップサート書き込みジョブのチェックポイント間隔は、3 分以上に設定してください。間隔が短いと、書き込み効率が低下し、多数の小規模ファイルが作成される可能性があります。

• 次の表は、Realtime Compute for Apache Flink と MaxCompute の間のデータ型のマッピングを示しています。

  Flink データ型	MaxCompute データ型
  CHAR(p)	CHAR(p)
  VARCHAR(p)	VARCHAR(p)
  STRING	STRING
  BOOLEAN	BOOLEAN
  TINYINT	TINYINT
  SMALLINT	SMALLINT
  INT	INT
  BIGINT	BIGINT
  FLOAT	FLOAT
  DOUBLE	DOUBLE
  DECIMAL(p, s)	DECIMAL(p, s)
  DATE	DATE
  TIMESTAMP(9) WITHOUT TIME ZONE, TIMESTAMP_LTZ(9)	TIMESTAMP
  TIMESTAMP(3) WITHOUT TIME ZONE, TIMESTAMP_LTZ(3)	DATETIME
  BYTES	BINARY
  ARRAY<T>	ARRAY<T>
  MAP<K, V>	MAP<K, V>
  ROW	STRUCT

  説明

  Flink の TIMESTAMP データ型にはタイムゾーン情報が含まれていませんが、MaxCompute の TIMESTAMP データ型には含まれています。この違いにより、8 時間の時間のずれが生じる可能性があります。タイムスタンプを揃えるには、TIMESTAMP_LTZ(9) を使用してください。

  -- Flink SQL
  CREATE TEMPORARY TABLE odps_source(
    id BIGINT NOT NULL COMMENT 'ID',
    created_time TIMESTAMP NOT NULL COMMENT '作成時間',
    updated_time TIMESTAMP_LTZ(9) NOT NULL COMMENT '更新時間',
  PRIMARY KEY (id) NOT ENFORCED
  ) WITH (
  'connector' = 'maxcompute',
  ...
  );

自己管理 Flink クラスターからのデータ書き込み

1. 前提条件： MaxCompute テーブルの作成

   まず、Flink がデータを書き込むための MaxCompute テーブルを作成する必要があります。次の例では、2 つのテーブル (非パーティション化 Delta テーブルとパーティションテーブル) を作成することで、このプロセスを示しています。テーブルプロパティの設定については、「Delta テーブルのパラメーター」をご参照ください。

   -- 非パーティション化 Delta テーブルを作成します。
   CREATE TABLE mf_flink_tt (
     id BIGINT not null,
     name STRING,
     age INT,
     status BOOLEAN, primary key (id)
   )
   tblproperties ("transactional"="true", 
                  "write.bucket.num" = "64", 
                  "acid.data.retain.hours"="12") ;
   
   --パーティション化 Delta テーブルを作成します。
   CREATE TABLE mf_flink_tt_part (
     id BIGINT not null,
     name STRING,
     age INT,
     status BOOLEAN, 
     primary key (id)
   )
     partitioned by (dd string, hh string) 
     tblproperties ("transactional"="true", 
                    "write.bucket.num" = "64", 
                    "acid.data.retain.hours"="12") ;
   

2. オープンソース Flink クラスターをセットアップします。コネクタは Flink 1.13、1.15、1.16、および 1.17 をサポートしています。ご利用の Flink バージョンに一致する Flink コネクタを選択してください。

   • Flink 1.13 用 Flink コネクタ

   • Flink 1.15 用 Flink コネクタ

   • Flink 1.16 および 1.17 用 Flink コネクタ

   説明

   • Flink 1.16 用の Flink コネクタは Flink 1.17 と互換性があります。

   • このトピックでは、Flink 1.13 用の Flink コネクタを例として使用します。パッケージをダウンロードして解凍してください。

3. Flink コネクタをダウンロードし、Flink クラスターパッケージに追加します。

   1. Flink コネクタの JAR パッケージをローカル環境にダウンロードします。

   2. 解凍した Flink インストールパッケージの lib ディレクトリに Flink コネクタの JAR パッケージを追加します。

      mv flink-connector-odps-1.13-shaded.jar $FLINK_HOME/lib/flink-connector-odps-1.13-shaded.jar

4. Flink インスタンスサービスを開始します。

   cd $FLINK_HOME/bin
   ./start-cluster.sh

5. Flink クライアントを開始します。

   cd $FLINK_HOME/bin
   ./sql-client.sh

6. Flink テーブルを作成し、Flink コネクタのパラメーターを設定します。

   Flink SQL または DataStream API のいずれかを使用して、Flink テーブルを作成し、そのパラメーターを設定できます。次のセクションでは、両方のアプローチの主要な例を示します。

   Flink SQL

   1. Flink SQL エディターで、次のコマンドを実行してテーブルを作成し、パラメーターを設定します。

      -- Flink SQL に非パーティション化テーブルを登録します。
      CREATE TABLE mf_flink (
        id BIGINT,
        name STRING,
        age INT,
        status BOOLEAN,
        PRIMARY KEY(id) NOT ENFORCED
      ) WITH (
        'connector' = 'maxcompute',
        'table.name' = 'mf_flink_tt',
        'sink.operation' = 'upsert',
        'odps.access.id'='LTAI****************',
        'odps.access.key'='********************',
        'odps.end.point'='http://service.cn-beijing.maxcompute.aliyun.com/api',
        'odps.project.name'='mf_mc_bj'
      );
      
      -- Flink SQL にパーティション化テーブルを登録します。
      CREATE TABLE mf_flink_part (
        id BIGINT,
        name STRING,
        age INT,
        status BOOLEAN,
        dd STRING,
        hh STRING,
        PRIMARY KEY(id) NOT ENFORCED
      ) PARTITIONED BY (`dd`,`hh`)
      WITH (
        'connector' = 'maxcompute',
        'table.name' = 'mf_flink_tt_part',
        'sink.operation' = 'upsert',
        'odps.access.id'='LTAI****************',
        'odps.access.key'='********************',
        'odps.end.point'='http://service.cn-beijing.maxcompute.aliyun.com/api',
        'odps.project.name'='mf_mc_bj'
      );

   2. Flink テーブルにデータを書き込み、MaxCompute テーブルをクエリして結果を確認します。

      -- Flink SQL クライアントで非パーティション化テーブルにデータを挿入します。
      INSERT INTO mf_flink VALUES (1,'Danny',27, false);
      
      -- MaxCompute でのクエリ結果。
      SELECT * FROM mf_flink_tt;
      +------------+------+------+--------+
      | id         | name | age  | status |
      +------------+------+------+--------+
      | 1          | Danny | 27   | false  |
      +------------+------+------+--------+
      
      -- Flink SQL クライアントで非パーティション化テーブルにデータを挿入してレコードを更新します。
      INSERT INTO mf_flink VALUES (1,'Danny',28, false);
      -- MaxCompute でのクエリ結果。
      SELECT * FROM mf_flink_tt;
      +------------+------+------+--------+
      | id         | name | age  | status |
      +------------+------+------+--------+
      | 1          | Danny | 28   | false  |
      +------------+------+------+--------+
      
      -- Flink SQL クライアントでパーティション化テーブルにデータを挿入します。
      INSERT INTO mf_flink_part VALUES (1,'Danny',27, false, '01','01');
      -- MaxCompute でのクエリ結果。
      SELECT * FROM mf_flink_tt_part WHERE dd=01 AND hh=01;
      +------------+------+------+--------+----+----+
      | id         | name | age  | status | dd | hh |
      +------------+------+------+--------+----+----+
      | 1          | Danny | 27   | false  | 01 | 01 |
      +------------+------+------+--------+----+----+
      
      -- Flink SQL クライアントでパーティション化テーブルにデータを挿入してレコードを更新します。
      INSERT INTO mf_flink_part VALUES (1,'Danny',30, false, '01','01');
      -- MaxCompute でのクエリ結果。
      SELECT * FROM mf_flink_tt_part WHERE dd=01 AND hh=01;
      +------------+------+------+--------+----+----+
      | id         | name | age  | status | dd | hh |
      +------------+------+------+--------+----+----+
      | 1          | Danny | 30   | false  | 01 | 01 |
      +------------+------+------+--------+----+----+

   DataStream API

   1. DataStream API を使用するには、次の依存関係を追加します。

      <dependency>
        <groupId>com.aliyun.odps</groupId>
        <artifactId>flink-connector-maxcompute</artifactId>
                  <version>xxx</version>
                  <scope>system</scope>
                  <systemPath>${mvn_project.basedir}/lib/flink-connector-maxcompute-xxx-shaded.jar</systemPath>
      </dependency>

      説明

      「xxx」を実際のバージョン番号に置き換えてください。

   2. 次のサンプルコードは、テーブルを作成し、パラメーターを設定する方法を示しています。

      package com.aliyun.odps.flink.examples;
      
      import org.apache.flink.configuration.Configuration;
      import org.apache.flink.odps.table.OdpsOptions;
      import org.apache.flink.odps.util.OdpsConf;
      import org.apache.flink.odps.util.OdpsPipeline;
      import org.apache.flink.streaming.api.datastream.DataStream;
      import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
      import org.apache.flink.table.api.Table;
      import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
      import org.apache.flink.table.data.RowData;
      
      public class Examples {
      
          public static void main(String[] args) throws Exception {
              StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
              env.enableCheckpointing(120 * 1000);
      
              StreamTableEnvironment streamTableEnvironment = StreamTableEnvironment.create(env);
      
              Table source = streamTableEnvironment.sqlQuery("SELECT * FROM source_table");
              DataStream<RowData> input = streamTableEnvironment.toAppendStream(source, RowData.class);
      
              Configuration config = new Configuration();
              config.set(OdpsOptions.SINK_OPERATION, "upsert");
              config.set(OdpsOptions.UPSERT_COMMIT_THREAD_NUM, 8);
              config.set(OdpsOptions.UPSERT_MAJOR_COMPACT_MIN_COMMITS, 100);
      
              OdpsConf odpsConfig = new OdpsConf("accessid",
                      "accesskey",
                      "endpoint",
                      "project",
                      "tunnel endpoint");
      
              OdpsPipeline.Builder builder = OdpsPipeline.builder();
              builder.projectName("sql2_isolation_2a")
                      .tableName("user_ledger_portfolio")
                      .partition("")
                      .configuration(config)
                      .odpsConf(odpsConfig)
                      .sink(input, false);
              env.execute();
          }
      }

Flink 完全管理からのデータ書き込み

1. 前提条件： MaxCompute テーブルの作成

   Flink データ用のターゲット MaxCompute テーブルを作成する必要があります。次の例では、Delta テーブルを作成する方法を示しています。

   SET odps.sql.type.system.odps2=true;
   DROP TABLE mf_flink_upsert;
   CREATE TABLE mf_flink_upsert (
     c1 int not null, 
     c2 string, 
     gt timestamp,
     primary key (c1)
   ) 
     PARTITIONED BY (ds string)
     tblproperties ("transactional"="true",
                    "write.bucket.num" = "64", 
                    "acid.data.retain.hours"="12") ;

2. Flink コネクタは Flink 完全管理にプリロードされているため、手動でのインストールは不要です。コネクタの詳細は、Realtime Compute for Apache Flink コンソールで確認できます。

3. Flink テーブルを作成し、Flink SQL ジョブを使用してリアルタイムの Flink データを構築し、開発後にジョブをデプロイします。

   Flink ジョブ開発ページで、SQL ジョブを作成および編集します。次の例では、ランダムデータを生成するソーステーブル、MaxCompute に接続する結果テーブル、およびデータを転送するための INSERT 文を定義しています。SQL ジョブの開発方法の詳細については、「ジョブ開発マップ」をご参照ください。

   -- Flink ソーステーブルを作成します。
   CREATE TEMPORARY TABLE fake_src_table
   (
       c1 int,
       c2 VARCHAR,
       gt AS CURRENT_TIMESTAMP
   ) WITH (
     'connector' = 'faker',
     'fields.c2.expression' = '#{superhero.name}',
     'rows-per-second' = '100',
     'fields.c1.expression' = '#{number.numberBetween ''0'',''1000''}'
   );
   
   -- Flink に一時的な結果テーブルを作成します。
   CREATE TEMPORARY TABLE test_c_d_g 
   (
       c1 int,
       c2 VARCHAR,
       gt TIMESTAMP,
       ds varchar,
       PRIMARY KEY(c1) NOT ENFORCED
    ) PARTITIONED BY(ds)
    WITH (
       		'connector' = 'maxcompute',
       		'table.name' = 'mf_flink_upsert',
       		'sink.operation' = 'upsert',
       		'odps.access.id'='LTAI****************',
       		'odps.access.key'='********************',
       		'odps.end.point'='http://service.cn-beijing.maxcompute.aliyun.com/api',
       		'odps.project.name'='mf_mc_bj',
       		'upsert.write.bucket.num'='64'
   );
   
   -- Flink 計算ロジック。
   INSERT INTO test_c_d_g
   SELECT  c1 AS c1,
           c2 AS c2,
           gt AS gt,
           date_format(gt, 'yyyyMMddHH') AS ds
   FROM    fake_src_table;

   パラメーター：

   odps.end.point：対応するリージョンの内部ネットワークエンドポイントを使用します。

   upsert.write.bucket.num：この値は、MaxCompute で作成された Delta テーブルのwrite.bucket.num プロパティの値と一致している必要があります。

4. MaxCompute テーブルをクエリして、データが書き込まれたことを確認します。

   SELECT * FROM mf_flink_upsert WHERE ds=2023061517;
   
   -- ソースデータはランダムに生成されるため、結果は異なる場合があります。
   +------+----+------+----+
   | c1   | c2 | gt   | ds |
   +------+----+------+----+
   | 0    | Skaar | 2023-06-16 01:59:41.116 | 2023061517 |
   | 21   | Supah Century | 2023-06-16 01:59:59.117 | 2023061517 |
   | 104  | Dark Gorilla Grodd | 2023-06-16 01:59:57.117 | 2023061517 |
   | 126  | Leader | 2023-06-16 01:59:39.116 | 2023061517 |
   

付録： Flink コネクタのパラメーター

• 基本パラメーター

  パラメーター	必須	デフォルト値	説明
  connector	はい	—	コネクタのタイプをMaxCompute に設定します。
  odps.project.name	はい	—	MaxCompute プロジェクト名。
  odps.access.id	はい	—	アカウントの AccessKey ID。「AccessKey ペアページ」をご参照ください。
  odps.access.key	はい	—	アカウントの AccessKey Secret。「AccessKey ペアページ」をご参照ください。
  odps.end.point	はい	—	MaxCompute エンドポイント。リージョンごとのエンドポイントのリストについては、「エンドポイント」をご参照ください。
  odps.tunnel.end.point	いいえ	—	Tunnel サービスのパブリックエンドポイント。デフォルトでは、リクエストは適切な Tunnel エンドポイントに自動的にルーティングされます。このパラメーターを設定すると、特定のエンドポイントが使用され、自動ルーティングが無効になります。 異なるリージョンとネットワークにおける Tunnel エンドポイントの詳細については、「エンドポイント」をご参照ください。
  odps.tunnel.quota.name	いいえ	—	MaxCompute へのアクセスに使用される Tunnel クォータの名前。
  table.name	はい	—	MaxCompute テーブルの名前を[project.][schema.]table の形式で指定します。
  odps.namespace.schema	いいえ	false	3層モデルを使用するかどうかを指定します。3層モデルの詳細については、「スキーマ操作」をご参照ください。
  sink.operation	はい	insert	書き込みタイプ。有効な値はinsert またはupsert です。 説明 アップサートモードは、MaxCompute Delta テーブルでのみサポートされます。
  sink.parallelism	いいえ	—	書き込みの並列度。設定されていない場合、デフォルトでアップストリームソースの並列度が使用されます。 説明 最適な書き込みパフォーマンスとシンクノードでのメモリ節約を最大化するために、テーブルプロパティwrite.bucket.num が設定値の整数倍であることを確認してください。
  sink.meta.cache.time	いいえ	400	メタデータキャッシュのサイズ。
  sink.meta.cache.expire.time	いいえ	1200	メタデータのキャッシュ有効期限 (秒単位)。
  sink.coordinator.enable	いいえ	true	コーディネーターモードを有効にするかどうかを指定します。

• パーティションパラメーター

  パラメーター	必須	デフォルト値	説明
  sink.partition	いいえ	—	データが書き込まれるパーティションの名前。 動的パーティション分割を使用する場合、このパラメーターは動的パーティションの親パーティションの名前を指定します。
  sink.partition.default-value	いいえ	__DEFAULT_PARTITION__	動的パーティション分割が使用される場合のデフォルトのパーティション名。
  sink.dynamic-partition.limit	いいえ	100	動的パーティション分割中に、1 つのチェックポイントで同時に書き込むことができるパーティションの最大数。 説明 この値を大幅に増やすと、シンクノードでメモリ不足 (OOM) エラーが発生する可能性があります。同時パーティションの数がこの制限を超えると、ジョブは失敗します。
  sink.group-partition.enable	いいえ	false	動的パーティション分割を使用する際に、パーティションでグループ化するかどうかを指定します。
  sink.partition.assigner.class	いいえ	—	PartitionAssigner の実装クラス。

• FileCached モードの書き込みパラメーター

  多数の動的パーティションを持つジョブには、ファイルキャッシュモードを使用します。次のパラメーターでこのモードを設定します。

  パラメーター	必須	デフォルト値	説明
  sink.file-cached.enable	いいえ	false	FileCached モードを有効にします。多数の動的パーティションを持つジョブに推奨されます。 false： FileCached モードは無効です。 true： FileCached モードは有効です。 説明 動的パーティションの数が多い場合は、ファイルキャッシュモードを使用できます。
  sink.file-cached.tmp.dirs	いいえ	./local	FileCached モードでのデフォルトのファイルキャッシュディレクトリ。
  sink.file-cached.writer.num	いいえ	16	FileCached モードでの単一タスクの同時データアップロードスレッド数。 説明 このパラメーターの値を大幅に増やさないでください。過剰な数のパーティションが同時に書き込まれると、OOM エラーが発生する可能性が高くなります。
  sink.bucket.check-interval	いいえ	60000	FileCached モードでファイルサイズをチェックする間隔。単位：ミリ秒 (ms)。
  sink.file-cached.rolling.max-size	いいえ	16 M	単一のキャッシュファイルの最大サイズ。 ファイルがこのサイズを超えると、アップロードされます。
  sink.file-cached.memory	いいえ	64 M	FileCached モードでファイル書き込みに使用されるオフヒープメモリの最大サイズ。
  sink.file-cached.memory.segment-size	いいえ	128 KB	FileCached モードでファイル書き込みに使用されるバッファーサイズ。
  sink.file-cached.flush.always	いいえ	true	FileCached モードでファイル書き込みにキャッシュを使用するかどうかを指定します。
  sink.file-cached.write.max-retries	いいえ	3	FileCached モードでデータをアップロードする際のリトライ回数。

• Insert またはUpsert の書き込みパラメーター

  Upsert 書き込みパラメーター

  パラメーター	必須	デフォルト値	説明
  upsert.writer.max-retries	いいえ	3	アップサートライターがバケットへのデータ書き込みに失敗した後のリトライ回数。
  upsert.writer.buffer-size	いいえ	64 MB	Flink での単一のアップサートライターのキャッシュサイズ。 説明 すべてのバケットのバッファーサイズの合計が事前設定されたしきい値に達すると、システムは自動的にフラッシュ操作をトリガーしてサーバーにデータを更新します。 アップサートライターは複数のバケットに同時にデータを書き込みます。書き込み効率を向上させるために、このパラメーターの値を増やすことを推奨します。 多数のパーティションにデータが書き込まれる場合、OOM エラーが発生する可能性があります。この場合、このパラメーターの値を減らすことができます。
  upsert.writer.bucket.buffer-size	いいえ	1 MB	Flink での単一バケットのキャッシュサイズ。Flink サーバーのメモリリソースが不足している場合は、このパラメーターの値を減らすことができます。
  upsert.write.bucket.num	はい	—	送信先テーブルのバケット数は、write.bucket.num の値と同じでなければなりません。
  upsert.write.slot-num	いいえ	1	単一セッションで使用される Tunnel スロットの数。
  upsert.commit.max-retries	いいえ	3	アップサートセッションコミットのリトライ回数。
  upsert.commit.thread-num	いいえ	16	アップサートセッションコミットの並列度。 この値を高く設定しすぎないでください。同時コミット数が多くなるとリソース消費が増加し、パフォーマンスの問題や過剰なリソース消費を引き起こす可能性があります。
  upsert.major-compact.min-commits	いいえ	100	メジャーコンパクションをトリガーするために必要な最小コミット数。
  upsert.commit.timeout	いいえ	600	アップサートセッションコミットのタイムアウト期間。単位：秒 (s)。
  upsert.major-compact.enable	いいえ	false	メジャーコンパクションを有効にするかどうかを指定します。
  upsert.flush.concurrent	いいえ	2	単一パーティション内で同時にデータを書き込むことができるバケットの最大数。 説明 バケット内のデータがフラッシュされると、Tunnel スロットが占有されます。

  説明

  アップサート書き込みの推奨パラメーター設定の詳細については、「アップサート書き込みの推奨パラメーター設定」をご参照ください。

  Insert 書き込みパラメーター

  パラメーター	必須	デフォルト値	説明
  insert.commit.thread-num	いいえ	16	コミットセッションの並列度。
  insert.arrow-writer.enable	いいえ	false	Arrow フォーマットを使用するかどうかを指定します。
  insert.arrow-writer.batch-size	いいえ	512	Arrow バッチ内の最大行数。
  insert.arrow-writer.flush-interval	いいえ	100000	ライターのフラッシュ間隔。単位：ミリ秒 (ms)。
  insert.writer.buffer-size	いいえ	64 MB	バッファー付きライターのキャッシュサイズ。