履歴追跡の有効化 (SCD タイプ 2)
適用対象 : API ベースのパイプライン作成
履歴追跡設定は、 slowly changing dimensions (SCD) 設定とも呼ばれ、時間の経過に伴うデータの変更を処理する方法を決定します。 履歴追跡をオフにすると (SCD タイプ 1)、古いレコードがソースで更新および削除されるときに上書きされます。ヒストリー・トラッキングをオンにして (SCD タイプ 2)、これらの変更のヒストリーを保守します。ソース内の表または列を削除しても、SCD タイプ 1 が選択されている場合でも、そのデータは宛先から削除されません。
たとえば、次のテーブルを取り込むとします。

また、アリスの好きな色が1月2日に紫色に変わるとしましょう。
履歴追跡がオフ (SCD タイプ 1) の場合、取り込み パイプラインの次回の実行では、宛先テーブルのその行が更新されます。

履歴追跡がオン (SCD タイプ 2) の場合、取り込み パイプラインは古い行を保持し、更新を新しい行として追加します。古い行を非アクティブとしてマークして、どの行が最新の状態であるかを把握できます。

すべてのコネクタが SCD タイプ 2 をサポートしているわけではありません。サポートされているコネクタの一覧については、「 機能の互換性」を参照してください。
例: Google アナリティクス
SCD タイプ 2 は、カーソル列として last_updated_date を使用する users テーブルと pseudonymous_users テーブルでサポートされています。追加のみのイベント レベルのテーブルではサポートされていません。
- Databricks Asset Bundles
 - Databricks notebook
 - Databricks CLI
 
デフォルトでは、ヒストリー・トラッキングはオフになっています (SCD タイプ 1)。次の YAML ファイルの例を使用して、バンドル内のこの設定を変更できます。
resources:
  pipelines:
    pipeline_ga4:
      name: <pipeline>
      catalog: <destination-catalog>
      schema: <destination-schema>
      ingestion_definition:
        connection_name: <connection>
        objects:
          - table:
              source_url: <project-id>
              source_schema: <property-name>
              destination_catalog: <destination-catalog>
              destination_schema: <destination-schema>
              table_configuration:
                scd_type: SCD_TYPE_2
デフォルトでは、ヒストリー・トラッキングはオフになっています (SCD タイプ 1)。ノートブックで次の Python コード例を使用して、この設定を変更できます。
pipeline_spec = """
{
  "name": "<pipeline>",
  "ingestion_definition": {
    "connection_name": "<connection>",
    "objects": [
      {
        "table": {
          "source_url": "<project-id>",
          "source_schema": "<property-name>",
          "destination_catalog": "<destination-catalog>",
          "destination_schema": "<destination-schema>",
          "table_configuration": {
            "scd_type": "SCD_TYPE_2",
          }
        }
      }
    ]
  }
}
"""
デフォルトでは、ヒストリー・トラッキングはオフになっています (SCD タイプ 1)。CLI コマンドで次の JSON 仕様例を使用して、この設定を変更できます。
{
  "resources": {
    "pipelines": {
      "pipeline_ga4": {
        "name": "<pipeline>",
        "catalog": "<destination-catalog>",
        "schema": "<destination-schema>",
        "ingestion_definition": {
          "connection_name": "<connection>",
          "objects": [
            {
              "table": {
                "source_url": "<project-id>",
                "source_schema": "<property-name>",
                "destination_catalog": "<destination-catalog>",
                "destination_schema": "<destination-schema>",
                "table_configuration": {
                  "scd_type": "SCD_TYPE_2"
                }
              }
            }
          ]
        }
      }
    }
  }
}
例: SQL Server
パイプライン構成で指定するシーケンス列 (たとえば、 last_updated、 modified_at、または version_number) によって、各行バージョンがアクティブだった期間が決まります (ターゲット表の __START_AT 列と __END_AT 列に記録されます)。
次の sequence_by 列タイプがサポートされています。
- タイムスタンプ
 - Date
 - Integer
 - Long
 - String
 
- Databricks Asset Bundles
 - Databricks notebook
 - Databricks CLI
 
デフォルトでは、ヒストリー・トラッキングはオフになっています (SCD タイプ 1)。バンドルでこの設定を変更するには、次の YAML ファイルの例を使用します。
resources:
  pipelines:
    pipeline_sqlserver:
      name: <pipeline>
      catalog: <destination-catalog>
      schema: <destination-schema>
      ingestion_definition:
        connection_name: <connection>
        objects:
          - table:
              source_catalog: <source-catalog>
              source_schema: <source-schema>
              source_table: <source-table>
              destination_catalog: <destination-catalog>
              destination_schema: <destination-schema>
              table_configuration:
                scd_type: SCD_TYPE_2
                sequence_by: <sequence-column>
デフォルトでは、ヒストリー・トラッキングはオフになっています (SCD タイプ 1)。この設定を変更するには、ノートブックで次の Python コード例を使用します。
pipeline_spec = """
{
  "name": "<pipeline>",
  "ingestion_definition": {
    "connection_name": "<connection>",
    "objects": [
      {
        "table": {
          "source_catalog": "<source-catalog>",
          "source_schema": "<source-schema>",
          "source_table": "<source-table>",
          "destination_catalog": "<destination-catalog>",
          "destination_schema": "<destination-schema>",
          "table_configuration": {
            "scd_type": "SCD_TYPE_2",
            "sequence_by": "<version-number>"
          }
        }
      }
    ]
  }
}
"""
デフォルトでは、ヒストリー・トラッキングはオフになっています (SCD タイプ 1)。この設定を変更するには、CLI コマンドで次の JSON 仕様の例を使用します。
{
  "resources": {
    "pipelines": {
      "pipeline_sqlserver": {
        "name": "<pipeline>",
        "catalog": "<destination-catalog>",
        "schema": "<destination-schema>",
        "ingestion_definition": {
          "connection_name": "<connection>",
          "objects": [
            {
              "table": {
                "source_catalog": "<source-catalog>",
                "source_schema": "<source-schema>",
                "source_table": "<source-table>",
                "destination_catalog": "<destination-catalog>",
                "destination_schema": "<destination-schema>",
                "table_configuration": {
                  "scd_type": "SCD_TYPE_2",
                  "sequence_by": "<version-number>"
                }
              }
            }
          ]
        }
      }
    }
  }
}
例: Workday
- Databricks Asset Bundles
 - Databricks notebook
 - Databricks CLI
 
デフォルトでは、ヒストリー・トラッキングはオフになっています (SCD タイプ 1)。次の例の YAML では、Databricks アセット バンドルを使用してこの設定を変更します。
resources:
  pipelines:
    pipeline_workday:
      name: <pipeline>
      catalog: <destination-catalog>
      schema: <destination-schema>
      ingestion_definition:
        connection_name: <connection>
        objects:
          - report:
              source_url: <report-url>
              destination_catalog: <destination-catalog>
              destination_schema: <destination-schema>
              table_configuration:
                scd_type: SCD_TYPE_2
pipeline_spec = """
{
  "name": "<pipeline>",
  "ingestion_definition": {
    "connection_name": "<connection>",
    "objects": [
      {
        "report": {
          "source_url": "<report-url>",
          "destination_catalog": "<destination-catalog>",
          "destination_schema": "<destination-schema>",
          "table_configuration": {
            "scd_type": "SCD_TYPE_2",
          }
        }
      }
    ]
  }
}
"""
{
  "resources": {
    "pipelines": {
      "pipeline_workday": {
        "name": "<pipeline>",
        "catalog": "<destination-catalog>",
        "schema": "<destination-schema>",
        "ingestion_definition": {
          "connection_name": "<connection>",
          "objects": [
            {
              "report": {
                "source_url": "<report-url>",
                "destination_catalog": "<destination-catalog>",
                "destination_schema": "<destination-schema>",
                "table_configuration": {
                  "scd_type": "SCD_TYPE_2"
                }
              }
            }
          ]
        }
      }
    }
  }
}
制限
完全な更新を実行すると、テーブル全体が置き換えられます。 これにより、以前の行バージョンがすべて削除されます。新しい履歴は、更新ポイントから開始して追跡されます。