MaxCompute:Flink を使用した Delta テーブルへのニアリアルタイムの部分列更新

操作プロセス

1. ユーザーが異なるプラットフォームで自分の名前と電話番号を更新します。 StreamA は名前の更新を受信し、StreamB は電話番号の更新を受信します。

2. StreamA と StreamB の両方が、ユーザー情報管理システムに更新を送信します。

最終結果

• 部分列更新なし: StreamB の更新が StreamA の後に到着して処理された場合、StreamA によって更新されたばかりの名前情報が上書きされ ( StreamB が行全体を更新する場合 )、名前が古い値に戻ります。

• 部分列更新あり:

  • StreamA が更新を実行する場合、連絡先情報列に影響を与えることなく、名前列のみを操作します。

  • StreamB が更新を実行する場合、個人情報列に影響を与えることなく、電話番号列のみを操作します。

  最終的な結果は、ユーザーの名前が最新の情報に更新され、電話番号も最新の情報に更新されることです。 これらの更新は干渉なしで独立して実行され、ユーザー情報の整合性と正確性が保証されます。

操作プロセス

1. StreamA は、INSERT INTO table (pk, age) VALUES (1, 3) ; のようなコマンドで、ユーザーの年齢のみを更新したいと考えています。

2. 同時に、StreamB は、INSERT INTO table (pk, sex) VALUES (1, 'male') ; のようなコマンドで、ユーザーの性別のみを更新したいと考えています。

最終結果

• 部分列更新なし: プライマリキー 1 のレコードが上記の更新コマンドを受信すると、更新されているフィールド以外のすべてのフィールドが NULL に設定されます。 これにより、元の有効なデータが失われます。

• 動的更新あり:

  • 挿入操作がトリガーされると、システムは一部のフィールドにのみデータが含まれていることを識別します。

  • 部分列更新メカニズムにより、データを持つこれらのフィールドのみが更新されます。

  • 同時に、挿入操作でデータが提供されていないフィールドは、元の値を維持します。

パラメーター

部分列更新モードを構成するために、MaxCompute は次の 2 つの構成パラメーターを導入しました。

動的部分列更新の構成例

動的部分列更新が有効になっているテーブルを作成します。 例:

CREATE TABLE partialtable (
  pk INT,
  c1 STRING, 
  c2 STRING, 
  c3 STRING,
  PRIMARY KEY(pk) NOT ENFORCED
) WITH (
  'connector' = 'maxcompute',
  'odps.end.point' = 'https://service.cn-hangzhou-vpc.maxcompute.aliyun-inc.com/api', //VPC ネットワーク接続
  'odps.project.name' = 'project_name',
  'odps.namespace.schema' = 'true', //3 層モデルをサポートします。
  'table.name' = 'project.schema.tablename',
  'sink.operation' = 'upsert',
  'upsert.write.bucket.num' = '1',
  'upsert.partial-column.enable' = 'true', 
  'odps.access.id' = 'yourAccessId',
  'odps.access.key' = 'yourAccessKey'
);

テーブルに対する後続の操作とその結果の例:

1. 最初の列がプライマリキーであるデータ [1,a, b, c] をテーブルに挿入します。 初期データは [1, a, b, c] です。

   INSERT INTO partialtable VALUES (1, 'a', 'b', 'c'); 

2. プライマリキー 1 のレコードの 2 番目の列 c2 のみを d に更新します。 更新後のデータは [1, a, d, c] です。

   INSERT INTO partialtable(pk, c2) VALUES (1, 'd'); 

3. プライマリキー 1 のレコードの 3 番目の列 c3 のみを e に更新します。 更新後のデータは [1, a, d, e] です。

   INSERT INTO partialtable(pk, c3) VALUES (1, 'e'); 

静的部分列更新の構成例

c2 列のみを更新するテーブルを作成します。 このテーブルに対する後続の操作は c2 列にのみ影響し、他の列は変更されません。 例:

CREATE TABLE PartialTable2 (
  pk INT,
  c1 STRING, 
  c2 STRING, 
  c3 STRING,
  PRIMARY KEY(pk) NOT ENFORCED
) WITH (
  'connector' = 'maxcompute',
  'odps.end.point' = 'https://service.cn-hangzhou-vpc.maxcompute.aliyun-inc.com/api', //VPC ネットワーク接続
  'odps.project.name' = 'project_name',
  'odps.namespace.schema' = 'true', //3 層モデルをサポートします。
  'table.name' = 'project.schema.tablename',
  'sink.operation' = 'upsert',
  'upsert.write.bucket.num' = '1',
  'upsert.partial-column.enable' = 'true', 
  'upsert.partial-column.name' = 'c2', // c2 列のみを更新するように指定します。
  'odps.access.id' = 'yourAccessId',
  'odps.access.key' = 'yourAccessKey'
);