メインコンテンツまでスキップ
に最終更新

AUTO CDC INTO (パイプライン)

AUTO CDC ... INTOステートメントを使用して、LakeFlow Pipelinesのチェンジデータキャプチャ (CDC) 機能を使用するフローを作成します。このステートメントは、CDCソースから変更を読み取り、ストリーミングターゲットに適用します。

構文​

CREATE OR REFRESH STREAMING TABLE table_name;

CREATE FLOW flow_name AS AUTO CDC [ONCE] INTO table_name
FROM source
KEYS (keys)
[IGNORE NULL UPDATES [ON {columnList | * EXCEPT (exceptColumnList)}]]
[APPLY AS DELETE WHEN condition]
[APPLY AS TRUNCATE WHEN condition]
SEQUENCE BY orderByColumn
[SYSTEM SEQUENCE BY systemOrderByColumn]
[COLUMNS {columnList | * EXCEPT (exceptColumnList)}]
[STORED AS {SCD TYPE 1 | SCD TYPE 2 | BITEMPORAL}]
[TRACK HISTORY ON {columnList | * EXCEPT (exceptColumnList)}]
[COLUMNS TO UPDATE columnName]

他のパイプライン クエリと同じCONSTRAINT句を使用して、ターゲットのデータ品質制約を定義します。「パイプラインの期待値によるデータ品質の管理」を参照してください。

INSERTとUPDATE イベントのデフォルトの動作は、ソースからCDCイベントを アップサート することです。指定されたキーに一致するターゲットテーブルの行を更新するか、ターゲットテーブルに一致するレコードが存在しない場合に新しい行を挿入します。DELETEイベントの処理はAPPLY AS DELETE WHEN条件で指定できます。

重要

変更を適用するターゲット ストリーミング テーブルを宣言する必要があります。 オプションで、ターゲット テーブルのスキーマを指定できます。SCD タイプ 2 テーブルの場合、ターゲット テーブルのスキーマを指定するときに、 sequence_byフィールドと同じデータ型の__START_AT列と__END_AT列も含める必要があります。

「AUTO CDC APIs : パイプラインによる変更データ キャプチャの簡素化」を参照してください。

パラメーター​

  • ONCE

    ONCEを指定すると、ターゲットテーブルに一度限りの挿入、つまりバックフィルが実行されます。フルリフレッシュの場合を除き、含まれるパイプラインが更新されても再実行されません。

    この句はオプションです。

  • flow_name

    作成するフローの名前。

  • source

    データのソース。ソースは ストリーミング ソースである必要があります。ストリーム キーワードを使用して、ストリーミング セマンティクスを使用してソースから読み取ります。 読み取り中に既存のレコードの変更または削除が検出されると、エラーがスローされます。静的ソースまたは追加専用のソースから読み取るのが最も安全です。変更コミットを含むデータを取り込むには、Python とskipChangeCommitsオプションを使用してエラーを処理できます。

    ストリーミング データの詳細については、 「パイプラインを使用したデータの変換」を参照してください。

  • KEYS

    ソース データ内の行を一意に識別する列または列の組み合わせ。これらの列の値は、ターゲット テーブル内の特定のレコードに適用される CDC イベントを識別するために使用されます。

    列の組み合わせを定義するには、列をコンマで区切ったリストを使用します。

    この条項は必須です。

  • IGNORE NULL UPDATES

    ターゲットカラムのサブセットを含む更新の取り込みを許可します。CDCイベントが既存の行と一致し、IGNORE NULL UPDATES が指定されている場合、null の値を持つ列はターゲット内の既存の値を保持します。これは、nullの値を持つネストされた列にも適用されます。

    部分的な更新には、どの列がnull値を無視するかを制御するON句を追加します：

    • IGNORE NULL UPDATES ON columnList: 入力値がnullの場合、リストされた列のみが既存の値を保持します。その他すべての列は明示的なnull値を適用します。
    • IGNORE NULL UPDATES ON * EXCEPT (exceptColumnList)：一覧にある列を除くすべての列は、入力値が null の場合に既存の値を保持します。一覧にある列には、明示的な null 値が適用されます。

    この句はオプションです。

    デフォルトでは、既存の列がnull値で上書きされます。

  • APPLY AS DELETE WHEN

    CDCイベントをアップサートではなくDELETEとして扱う必要がある場合を指定します。

    SCD タイプ 2 ソースの場合、順不同のデータを処理するために、削除された行は基になる Delta テーブルに廃棄石として一時的に保持され、これらの廃棄石を除外するビューがメタストアに作成されます。保持間隔は、 pipelines.cdc.tombstoneGCThresholdInSeconds テーブル プロパティで構成できます。

    この句はオプションです。

  • APPLY AS TRUNCATE WHEN

    CDCイベントを完全なテーブルTRUNCATEとして扱う必要がある場合を指定します。この句はターゲットテーブルの完全な切り捨てをトリガーするため、この機能が必要な特定のユースケースにのみ使用してください。

    APPLY AS TRUNCATE WHEN句は SCD タイプ 1 でのみサポートされます。SCD タイプ 2 は切り捨て操作をサポートしていません。

    この句はオプションです。

  • SEQUENCE BY

    ソース データ内の CDC イベントの論理順序を指定する列名。パイプライン処理では、このシーケンスを使用して、順序どおりに到着しない変更イベントを処理します。

    シーケンス処理に複数の列が必要な場合は、STRUCT 式を使用します。これは、最初の構造体フィールドで最初に並べ替えられ、次に同点の場合は2番目のフィールドで、以降同様です。

    指定された列は並べ替え可能なデータ型である必要があります。

    この条項は必須です。

  • SYSTEM SEQUENCE BY

備考

ベータ版

バイテンポラル AUTO CDC はベータ版です。

各CDCイベントがシステムに認識されたシステム時間を指定する列名。STORED AS BITEMPORAL と共に使用し、ビジネス時間 (SEQUENCE BY) とシステム時間の両方で変更を追跡します。「バイテンポラル AUTO CDC」を参照してください。

指定された列は並べ替え可能なデータ型である必要があります。

この句はオプションであり、バイテンポラルテーブルにのみ適用されます。

  • COLUMNS

    ターゲット テーブルに含める列のサブセットを指定します。次のいずれかを実行できます。

    • 含める列の完全なリストを指定します: COLUMNS (userId, name, city) 。
    • 除外する列のリストを指定します: COLUMNS * EXCEPT (operation, sequenceNum)

    この句はオプションです。

    COLUMNS句が指定されていない場合、デフォルトではターゲット テーブルのすべての列が含まれます。

  • STORED AS

    レコードをSCDタイプ1、SCDタイプ2、またはバイテンポラルとして保存するかどうか。

    ビジネス時間とシステム時間の両方で変更を追跡するには、BITEMPORAL に設定します。Bitemporal には SYSTEM SEQUENCE BY が必要で、ベータ版です。Bitemporal AUTO CDC を参照してください。

    この句はオプションです。

    デフォルトはSCDタイプ1です。

  • TRACK HISTORY ON

    指定された列に変更があった場合に履歴レコードを生成する出力列のサブセットを指定します。次のいずれかを実行できます。

    • 追跡する列の完全なリストを指定します: COLUMNS (userId, name, city) 。
    • 追跡から除外する列のリストを指定します。 COLUMNS * EXCEPT (operation, sequenceNum)

    この条項はオプションです。デフォルトでは、変更があった場合にすべての出力列の履歴を追跡します。これはTRACK HISTORY ON *と同じです。

  • COLUMNS TO UPDATE

    変更レコードごとに、更新する列のセットを列名の文字列の配列として保持するソース列の名前を指定します（array<string>）。配列にない列は既存のターゲット値を保持し、リストされている列は明示的なnull値を含むソースから書き込まれます。

    各変更レコードが異なる列セットを更新し、明示的な null の値を適用する必要がある場合は、この句を部分更新に使用します。

    COLUMNS TO UPDATE を IGNORE NULL UPDATES と一緒に使用することはできません。また、バイテンポラルテーブルではサポートされていません。

    この句はオプションです。

例​

SQL
-- Create a streaming table, then use AUTO CDC to populate it:
CREATE OR REFRESH STREAMING TABLE target;

CREATE FLOW flow
AS AUTO CDC INTO
  target
FROM stream(cdc_data.users)
  KEYS (userId)
  APPLY AS DELETE WHEN operation = "DELETE"
  SEQUENCE BY sequenceNum
  COLUMNS * EXCEPT (operation, sequenceNum)
  STORED AS SCD TYPE 2
  TRACK HISTORY ON * EXCEPT (city);
このページの見出し