Data Warehouse からレイクハウスにデータを増分読み込みする

[アーティクル]
11/15/2023

このチュートリアルでは、Data Warehouse からレイクハウスにデータを増分読み込む方法について説明します。

概要

ソリューションの概略図を次に示します。

Diagram showing incrementally load data logic.

このソリューションを作成するための重要な手順を次に示します。

基準値列を選択する。ソースデータテーブルのいずれか 1 つの列を選択します。実行ごとに新しいレコードまたは更新されたレコードを切り分ける目的で使用されます。通常、行が作成または更新されたときに常にデータが増える列を選択します (last_modify_time、ID など)。この列の最大値が基準値として使用されます。
Data Warehouse に最後のウォーターマークの値を格納するテーブルを準備します。
次のワークフローを含んだパイプラインを作成する。

このソリューションのパイプラインには、次のアクティビティがあります。
- 2 つのルックアップアクティビティを作成する。 1 つ目のルックアップアクティビティは、前回の基準値を取得するために使用します。 2 つ目のルックアップアクティビティは、新しい基準値を取得するために使用します。これらの基準値は、コピーアクティビティに渡されます。
- ウォーターマーク列の値が古いウォーターマーク値より大きく、新しいウォーターマーク値より小さいソースデータテーブルから行をコピーするコピーアクティビティを作成します。次に、Data Warehouse からレイクハウスに新しいファイルとしてデータをコピーします。
- ストアドプロシージャアクティビティを作成する。これはパイプラインの次回実行に備えて基準値を更新するためのアクティビティです。

前提条件

Data Warehouse。 Data Warehouse をソースデータストアとして使用します。ない場合には、「Data Warehouse の作成」で作成する手順を参照してください。
レイクハウス。レイクハウスをコピー先データストアとして使用します。ない場合には、「レイクハウスの作成」で作成する手順を参照してください。コピーしたデータを格納する IncrementalCopy という名前のフォルダーを作成します。

ソースの準備

増分コピーパイプラインを構成する前に、ソース Data Warehouse で準備する必要があるテーブルとストアドプロシージャをいくつか次に示します。

1. Data Warehouse にデータソーステーブルを作成する

Data Warehouse で次の SQL コマンドを実行し、データソーステーブルとして data_source_table という名前のテーブルを作成します。このチュートリアルでは、増分コピーを実行するためのサンプルデータとして使用します。

create table data_source_table
(
    PersonID int,
    Name varchar(255),
    LastModifytime DATETIME2(6)
);

INSERT INTO data_source_table
    (PersonID, Name, LastModifytime)
VALUES
    (1, 'aaaa','9/1/2017 12:56:00 AM'),
    (2, 'bbbb','9/2/2017 5:23:00 AM'),
    (3, 'cccc','9/3/2017 2:36:00 AM'),
    (4, 'dddd','9/4/2017 3:21:00 AM'),
    (5, 'eeee','9/5/2017 8:06:00 AM');

データソーステーブル内のデータを以下に示します:

PersonID | Name | LastModifytime
-------- | ---- | --------------
1        | aaaa | 2017-09-01 00:56:00.000
2        | bbbb | 2017-09-02 05:23:00.000
3        | cccc | 2017-09-03 02:36:00.000
4        | dddd | 2017-09-04 03:21:00.000
5        | eeee | 2017-09-05 08:06:00.000

このチュートリアルでは、LastModifytime を基準値列として使用します。

2. Data Warehouse に最後のウォーターマークの値を格納するテーブルを準備する

Data Warehouse で次の SQL コマンドを実行し、watermarktable という名前のテーブルを作成し、最後のウォーターマーク値を格納します:
```
create table watermarktable
(
TableName varchar(255),
WatermarkValue DATETIME2(6),
);
```
最後のウォーターマークの既定値をソースデータテーブルのテーブル名で設定します。このチュートリアルでは、テーブル名は data_source_table、既定値は 1/1/2010 12:00:00 AM です。
```
INSERT INTO watermarktable
VALUES ('data_source_table','1/1/2010 12:00:00 AM')    
```

テーブルの watermarktable のデータを確認します。

Select * from watermarktable

出力:

TableName  | WatermarkValue
----------  | --------------
data_source_table | 2010-01-01 00:00:00.000

3. Data Warehouse にストアドプロシージャを作成する

次のコマンドを実行して、Data Warehouse にストアドプロシージャを作成します。このストアドプロシージャは、最後のパイプライン実行後に最後の基準値を更新するために使用されます。

CREATE PROCEDURE usp_write_watermark @LastModifiedtime datetime, @TableName varchar(50)
AS

BEGIN

UPDATE watermarktable
SET [WatermarkValue] = @LastModifiedtime
WHERE [TableName] = @TableName

END

増分コピー用のパイプラインを構成する

手順 1: パイプラインを作成する

Power BI に移動します。
画面の左下にある Power BI アイコンを選択し、[データファクトリ] を選択して Data Factory のホームページを開きます。
Microsoft Fabric ワークスペースに移動します。
[データパイプライン] を選択し、パイプライン名を入力して新しいパイプラインを作成します。

手順 2: 最後のウォーターマークの検索回数アクティビティを追加する

この手順では、最後の基準値を取得する検索回数アクティビティを作成します。前に設定された既定値 1/1/2010 12:00:00 AM が取得されます。

[パイプラインアクティビティの追加] を選択し、ドロップダウンリストから [参照] を選択します。
[全般] タブで、このアクティビティの名前を LookupOldWaterMarkActivity に変更します。
[設定] タブで、次の構成を実行します:
- [データストアの種類]: [ワークスペース] を選択します。
- [ワークスペースデータの種類]: [Data Warehouse] を選択します。
- [Data Warehouse]: [Data Warehouse] を選択します。
- [クエリの使用]: [テーブル] を選択します。
- [テーブル]: [dbo.watermarktable] を選択します。
- [最初の行のみ]: 選択済みです。

手順 3: 新しいウォーターマークの検索回数アクティビティを追加する

この手順では、新しい基準値を取得する検索回数アクティビティを作成します。クエリを使用して、ソースデータテーブルから新しいウォーターマークを取得します。 data_source_tableの LastModifytime 列の最大値が取得されます。

上部のバーで、[アクティビティ] タブの [検索回数] を選択して、2 番目の検索回数アクティビティを追加します。
[全般] タブで、このアクティビティの名前を LookupOldWaterMarkActivity に変更します。
[設定] タブで、次の構成を実行します:
- [データストアの種類]: [ワークスペース] を選択します。
- [ワークスペースデータの種類]: [Data Warehouse] を選択します。
- [Data Warehouse]: [Data Warehouse] を選択します。
- [クエリの使用]: [クエリ] を選択します。
- [クエリ]: 次のクエリを入力して、最後に変更された最大時間を新しい基準値として選択します:
```
select MAX(LastModifytime) as NewWatermarkvalue from data_source_table
```
- [最初の行のみ]: 選択済みです。

手順 4: 増分データをコピーするコピーアクティビティを追加する

この手順では、コピーアクティビティを追加して、最後のウォーターマークと新しいウォーターマークの間の増分データを Data Warehouse からレイクハウスにコピーします。

上部のバーで [アクティビティ] を選択し、[データのコピー] ->[キャンバスに追加] を選択してコピーアクティビティを取得します。
[全般] タブで、このアクティビティの名前を IncrementalCopyActivity に変更します。
2 つの検索回数アクティビティにコピーアクティビティに接続します。これには、検索回数アクティビティにアタッチされている緑のボタン (成功時) をコピーアクティビティにドラッグします。コピーアクティビティの境界線の色が緑に変わったら、マウスボタンを離します。
[ソース] タブで、次の構成を実行します:
- [データストアの種類]: [ワークスペース] を選択します。
- [ワークスペースデータの種類]: [Data Warehouse] を選択します。
- [Data Warehouse]: [Data Warehouse] を選択します。
- [クエリの使用]: [クエリ] を選択します。
- クエリ: 次のクエリを入力して、最後のウォーターマークと新しいウォーターマークの間に増分データをコピーします。
```
select * from data_source_table where LastModifytime > '@{activity('LookupOldWaterMarkActivity').output.firstRow.WatermarkValue}' and LastModifytime <= '@{activity('LookupNewWaterMarkActivity').output.firstRow.NewWatermarkvalue}'
```
[コピー先] タブで、次の構成を実行します:
- [データストアの種類]: [ワークスペース] を選択します。
- [ワークスペースデータストアの種類]: [レイクハウス] を選択します。
- [レイクハウス]: [レイクハウス] を選択します。
- [ルートフォルダー]: [ファイル] を選択します。
- [ファイルパス]: コピーしたデータを格納するフォルダーを指定します。 [参照] を選択してフォルダーを選択します。ファイル名として、[動的コンテンツの追加] を開き、開いたウィンドウに「@CONCAT('Incremental-', pipeline().RunId, '.txt')」と入力して、レイクハウスでコピーしたデータファイルのファイル名を作成します。
- [ファイル形式]: データの形式の種類を選択します。

手順 5: ストアドプロシージャアクティビティを追加する

この手順では、ストアドプロシージャアクティビティを追加して、次のパイプライン実行の最後のウォーターマーク値を更新します。

上部のバーで [アクティビティ] を選択し、[ストアドプロシージャ] を選択してストアドプロシージャアクティビティを追加します。
[全般] タブで、このアクティビティの名前を StoredProceduretoWriteWatermarkActivity に変更します。
コピーアクティビティの緑の (成功時) 出力をストアドプロシージャアクティビティに接続します。

[設定] タブで、次の構成を実行します:

[データストアの種類]: [ワークスペース] を選択します。
[Data Warehouse]: [Data Warehouse] を選択します。
[ストアドプロシージャ名]: Data Warehouse で作成したストアドプロシージャを指定します: [dbo].[usp_write_watermark]。

ストアドプロシージャパラメータを展開します。ストアドプロシージャのパラメーターの値を指定するには、[インポート] を選択し、各パラメーターに次の値を入力します:

名前	Type	値
LastModifiedtime	DateTime	@{activity('LookupNewWaterMarkActivity').output.firstRow.NewWatermarkvalue}
TableName	String	@{activity('LookupOldWaterMarkActivity').output.firstRow.TableName}

Screenshot showing stored procedure activity configuration.

手順 6:パイプラインを実行し、結果を監視する

上部のバーで、[ホーム] タブの [実行] を選択します。次に、[保存して実行] を選択します。パイプライン実行が開始され、[出力] タブでパイプラインを監視できます。

Screenshot showing pipeline run results.

レイクハウスに移動すると、指定したフォルダーの下にデータファイルが見つかります。ファイルを選択して、コピーしたデータをプレビューできます。

Screenshot showing lakehouse data for the first pipeline run.

Screenshot showing lakehouse data preview for the first pipeline run.

さらにデータを追加して増分コピーの結果を表示する

最初のパイプラインの実行が完了したら、Data Warehouse ソーステーブルにさらにデータを追加して、このパイプラインで増分データをコピーできるかどうかを確認してみましょう。

手順 1: ソースにデータを追加する

次のクエリを実行して、Data Warehouse に新しいデータを挿入します:

INSERT INTO data_source_table
VALUES (6, 'newdata','9/6/2017 2:23:00 AM')

INSERT INTO data_source_table
VALUES (7, 'newdata','9/7/2017 9:01:00 AM')

data_source_table の更新されたデータは次のとおりです:

PersonID | Name | LastModifytime
-------- | ---- | --------------
1 | aaaa | 2017-09-01 00:56:00.000
2 | bbbb | 2017-09-02 05:23:00.000
3 | cccc | 2017-09-03 02:36:00.000
4 | dddd | 2017-09-04 03:21:00.000
5 | eeee | 2017-09-05 08:06:00.000
6 | newdata | 2017-09-06 02:23:00.000
7 | newdata | 2017-09-07 09:01:00.000

手順 2: 別のパイプライン実行をトリガーし、結果を監視する

パイプラインページに戻ります。上部のバーで、[ホーム] タブの [実行] をもう一度選択します。パイプライン実行が開始され、[出力] タブでパイプラインを監視できます。

レイクハウスに移動すると、新しいコピーされたデータファイルが指定したフォルダーの下にあります。そのファイルを選択してコピーされたデータをプレビューすることができます。このファイルに増分データが表示されます。

Screenshot showing lakehouse data for the second pipeline run.

Screenshot showing lakehouse data preview for the second pipeline run.

次に、Azure Blob Storage からレイクハウスへのコピーの詳細に進みます。

Azure Blob Storage からレイクハウスへのコピー

次の方法で共有

Data Warehouse からレイクハウスにデータを増分読み込みする

概要

前提条件

ソースの準備

1. Data Warehouse にデータソーステーブルを作成する

2. Data Warehouse に最後のウォーターマークの値を格納するテーブルを準備する

3. Data Warehouse にストアドプロシージャを作成する

増分コピー用のパイプラインを構成する

手順 1: パイプラインを作成する

手順 2: 最後のウォーターマークの検索回数アクティビティを追加する

手順 3: 新しいウォーターマークの検索回数アクティビティを追加する

手順 4: 増分データをコピーするコピーアクティビティを追加する

手順 5: ストアドプロシージャアクティビティを追加する

手順 6:パイプラインを実行し、結果を監視する

さらにデータを追加して増分コピーの結果を表示する

手順 1: ソースにデータを追加する

手順 2: 別のパイプライン実行をトリガーし、結果を監視する

フィードバック

その他のリソース

次の方法で共有

Data Warehouse からレイクハウスにデータを増分読み込みする

概要

前提条件

ソースの準備

1. Data Warehouse にデータ ソース テーブルを作成する

2. Data Warehouse に最後のウォーターマークの値を格納するテーブルを準備する

3. Data Warehouse にストアド プロシージャを作成する

増分コピー用のパイプラインを構成する

手順 1: パイプラインを作成する

手順 2: 最後のウォーターマークの検索回数アクティビティを追加する

手順 3: 新しいウォーターマークの検索回数アクティビティを追加する

手順 4: 増分データをコピーするコピー アクティビティを追加する

手順 5: ストアド プロシージャ アクティビティを追加する

手順 6:パイプラインを実行し、結果を監視する

さらにデータを追加して増分コピーの結果を表示する

手順 1: ソースにデータを追加する

手順 2: 別のパイプライン実行をトリガーし、結果を監視する

関連するコンテンツ

フィードバック

その他のリソース

1. Data Warehouse にデータソーステーブルを作成する

3. Data Warehouse にストアドプロシージャを作成する

手順 4: 増分データをコピーするコピーアクティビティを追加する

手順 5: ストアドプロシージャアクティビティを追加する