Azure Data Factory メタデータ駆動のコピータスクにより ingest したファイルを Databricks で取り込む方法

## 概要

Azure Data Factory メタデータ駆動のコピータスクにより ingest したファイルを Databricks で取り込む方法を共有します。Azure Data Factory メタデータ駆動のコピータスクを組み合わせことで、大量のソースデータをレイクハウスアーキテクチャに高速、かつ、効率よく取り込むことができます。メタデータ駆動のコピータスクのメイン制御テーブルに Databricks の実行に必要となるパラメータを渡すことを想定しています。

```sql
    [DatabricksLoadingBehaviorSettings] [nvarchar](max) NULL,
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/c044eeca-6f55-2997-4db9-46244e281d8a.png)

```json
        "DatabricksLoadingBehaviorSettings": {
            "src_dir": "abfss://metadata-test@metadata-test.dfs.core.windows.net/test06",
            "checkpoint_dir": "abfss://metadata-test@metadata-test.dfs.core.windows.net/checkpoint",
            "tgt_tbl_name": "hive_metastore.metadata_copy_01.table_01"
        },
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/14cb675b-2e41-0e33-fe51-8aa4eb782505.png)

本記事の実装方法は、投稿済みである下記記事の実装方法の応用です。

- [Databricks にて json 文字列をパラメータで渡すことによるノートブック実行処理の共通化 #Spark - Qiita](https://qiita.com/manabian/items/53f4461b31b1048dc894)

## 概要

### 1. Azure SQL Database 上にソースとなるテーブルを作成とデータ挿入

```sql
IF  OBJECT_ID(N'dbo.adf_metadata_01', N'U') IS NOT NULL
    DROP TABLE dbo.adf_metadata_01
CREATE TABLE dbo.adf_metadata_01
(
    ID INT IDENTITY(1,1) PRIMARY KEY,
	NAME varchar(10),
    update_timestamp DATETIME
);

INSERT INTO dbo.adf_metadata_01
(
	NAME,
	update_timestamp
)
SELECT 'A' AS NAME,CAST('2024-06-01' AS datetime2) AS update_timestamp
;

SELECT
	*
	FROM
		dbo.adf_metadata_01
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/523e9a67-fd63-36ad-f523-b7348786ac80.png)


### 2. Databricks に書き込み先のスキーマとテーブルを作成

```sql
%sql
CREATE SCHEMA IF NOT EXISTS hive_metastore.metadata_copy_01;
CREATE OR REPLACE TABLE hive_metastore.metadata_copy_01.table_01
(
    ID STRING,
  	NAME STRING,
    update_timestamp STRING
)
;
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/62c1fbec-c8e8-d68a-866f-3de6d4f2d2aa.png)

### 3. リンク先の記事を参考に Databricks Wokspace 上にノートブックを作成

- [Databricks にて json 文字列をパラメータで渡すことによるノートブック実行処理の共通化 #Spark - Qiita](https://qiita.com/manabian/items/53f4461b31b1048dc894)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/3c50d17d-ae80-6623-d58f-668ff976edab.png)


### 4. メタデータ駆動のコピータスクを作成

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/0c0958f2-1e00-679a-8b5c-cedef6b61fce.png)

### 5. メイン制御テーブルの一部を修正後に実行

```sql
    [DatabricksLoadingBehaviorSettings] [nvarchar](max) NULL,
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/c044eeca-6f55-2997-4db9-46244e281d8a.png)

```json
        "DatabricksLoadingBehaviorSettings": {
            "src_dir": "abfss://metadata-test@metadata-test.dfs.core.windows.net/test06",
            "checkpoint_dir": "abfss://metadata-test@metadata-test.dfs.core.windows.net/checkpoint",
            "tgt_tbl_name": "hive_metastore.metadata_copy_01.table_01"
        },
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/14cb675b-2e41-0e33-fe51-8aa4eb782505.png)

```sql
                [DatabricksLoadingBehaviorSettings],
```

```sql
                [DatabricksLoadingBehaviorSettings] [nvarchar](max) AS JSON,
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/78cb97ed-35e1-94f9-9c91-b3d3fdad888e.png)

上記の修正後に Azure SQL Database でコードを実行します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/6865c8de-3d09-82ce-c451-c8d225008cb4.png)

### 6. `MetadataDrivenCopyTask_xxx_BottomLevel`パイプラインにて作成したノートブックにベースパラメータとして`pipe_conf`という名前で`@item().DatabricksLoadingBehaviorSettings`を渡す Databricks ノートブックアクティビティを追加

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/ee97a948-1a4a-d882-a23a-55bc3450c82c.png)


### 7. Azure Data Factory にてメタデータ駆動のコピータスクをデバック実行し正常終了することを確認

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/26384e86-4872-a3c3-d60f-5ce7e13b45d6.png)

### 8. Databricks のテーブルにデータが書きこまれていることを確認

```python
spark.table("hive_metastore.metadata_copy_01.table_01").display()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/6ff708b1-1002-c89c-15bc-1d83cc5a13ec.png)


### 9. 作成したリソースを削除

```python
spark.sql("DROP SCHEMA IF EXISTS hive_metastore.metadata_copy_01 CASCADE")
dbutils.fs.rm("abfss://metadata-test@metadata-test.dfs.core.windows.net/checkpoint", True)
dbutils.fs.rm("abfss://metadata-test@metadata-test.dfs.core.windows.net/test06", True)
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/f3a8f082-5c07-0a96-73e6-3d61c652ff9c.png)