Azure Synapse Analytics の Mapping Data Flow で CSV を Parquet & Delta に変換する

# はじめに
GUI ベースで ETL 処理が行える **Mapping Data Flow** は Azure Data Factory だけでなく **Azure Synapse Analytics** で使うことができます。
Mapping Data Flow は元々 **Parquet** 形式をサポートしていましたが、最近 **Delta** 形式もサポート対象に加わりました。
また、Azure Synapse Analytics では **SQL オンデマンド** というサーバーレスなクエリ実行エンジンを利用できます。

本記事では、Azure Blob Storage に格納された CSV ファイルを Mapping Data Flow で Parquet 形式 & Delta 形式に変換して Azure Data Lake Storage Gen2 (ADLS) に格納し、SQL オンデマンドでクエリを実行するまでの一連の流れを簡単にまとめたいと思います。

# 目次
1. [Synapse Workspace 作成](https://qiita.com/nakazax/items/877bda4e74430d388a36#1-synapse-workspace-%E4%BD%9C%E6%88%90)
2. [入出力ファイル & フォルダ準備](https://qiita.com/nakazax/items/877bda4e74430d388a36#2-%E5%85%A5%E5%87%BA%E5%8A%9B%E3%83%95%E3%82%A1%E3%82%A4%E3%83%AB--%E3%83%95%E3%82%A9%E3%83%AB%E3%83%80%E6%BA%96%E5%82%99)
3. [Linked services & Datasets 設定](https://qiita.com/nakazax/items/877bda4e74430d388a36#3-linked-services--datasets-%E8%A8%AD%E5%AE%9A)
4. [Data flows 作成](https://qiita.com/nakazax/items/877bda4e74430d388a36#4-data-flows-%E4%BD%9C%E6%88%90)
5. [Pipelines 作成 & 実行](https://qiita.com/nakazax/items/877bda4e74430d388a36#5-pipelines-%E4%BD%9C%E6%88%90--%E5%AE%9F%E8%A1%8C)
6. [SQL オンデマンドでのクエリ実行](https://qiita.com/nakazax/items/877bda4e74430d388a36#6-sql-%E3%82%AA%E3%83%B3%E3%83%87%E3%83%9E%E3%83%B3%E3%83%89%E3%81%A7%E3%81%AE%E3%82%AF%E3%82%A8%E3%83%AA%E5%AE%9F%E8%A1%8C)

# 1. Synapse Workspace 作成
Azure ポータルで [synapse] と検索し [Azure Synapse Analytics (workspaces preview)] を選択します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/284925/a63f68dd-cfab-e6bc-5cac-f40e06b75070.png)

Workspace 作成画面です。20207/26 時点では東日本 & 西日本リージョンは選択できません。任意のリージョンを指定します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/284925/6ea91ebd-6094-6717-d174-dbb8b2e87cf2.png)

SQL オンデマンドで ADLS のデータに対してクエリを実行する際に権限的に必要なためチェックしておきます。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/284925/e27693a6-e88f-d371-8b29-80fc70e0f018.png)

次画面では SQL 管理者のユーザー名とパスワードを入力します。他のチェックボックスについては必要に応じてオン/オフを切り替えましょう。
特定の IP アドレスからのアクセスのみを許可したい場合は、一番下の [Allow connection from all IP address] のチェックを外しておきます。Workspace 作成後に Firewall 画面から接続を許可する IP アドレスを編集します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/284925/64379aa6-d44c-0c6a-1737-f617c56486a2.png)

問題なければ作成します。数分後にリソース作成が完了します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/284925/2e02a4cf-8f9c-eb20-f523-e744f5639b6d.png)

# 2. 入出力ファイル & フォルダ準備
入力用の準備として、ダミーデータを準備して Blob Storage にアップロードします。
出力用の準備として、ADLS にフォルダーを作成しておきます。

## 2-1. 入力用 - ダミーデータ作成
まずは Blob Storage に格納する CSV データを準備します。ここでは mockaroo というダミーデータ生成用の Web サイトでダミーデータを作成します。mockaroo では 1000 件 / 回のダミーデータを無料で作成できます。
https://www.mockaroo.com/

こんな感じでフィールド名と型とオプションを入れると良い感じのダミーデータが作成されます。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/284925/a76c6493-3b1c-30c6-968c-6153d385092f.png)

作成したデータは以下のような感じです。

```
order_number,created_at,customer_id,stock_name,quantity
1,2020-07-07T16:58:56Z,cb500cf5-4786-4bdc-b1a8-93652f31cfae,"American Eagle Outfitters, Inc.",82
2,2020-07-07T10:22:17Z,6c57e361-d511-4893-a2a1-af1579f3d15c,"Meridian Waste Solutions, Inc",24
...
999,2020-07-07T02:11:19Z,87f29c2d-30b0-4714-a2db-efe47e28831c,LKQ Corporation,45
1000,2020-07-05T04:34:07Z,58a4d463-9fc9-4f0c-9745-797e0427383f,Valvoline Inc.,29
```

## 2-2. 入力用 - ダミーデータを Blob Storage にアップロード
あらかじめ 作っておいた Blob Storage に Storage Explorer でダミーデータをアップロードします。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/284925/b0a24040-cffe-aded-7de0-70a41c40aa2a.png)

## 2-3. 出力用 - フォルダ作成
この後 Datasets を設定する際に ADLS にフォルダが存在する必要があるため、ADLS の Storage Explorer でフォルダを作成しておきます。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/284925/3d807c70-516e-e41f-374c-322eacb431f2.png)

# 3. Linked services & Datasets 設定
ここから先は Synapse Workspace で作業を行います。Synapse Workspace の Overview 画面の赤枠で囲んだリンクをクリックすると Workspace が別タブで起動します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/284925/a6e2f4cd-58eb-9acb-2f55-0864b743cdde.png)

## 3-1. Linked services - Blob
(Name: BlobStorageNakazaxQiitaDf)
入力 Blob Storage 用の Linked service です。
[Manage] ハブ > [Linked services] で [New] をクリックして新規 Linked service を作成します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/284925/91708194-d6c6-f3ea-6291-65ddc6c3e6bc.png)

## 3-2. Datasets  - Blob
(Name: OrdersCsv)
入力データ用の Dataset です。[Data] ハブの [+] アイコンをクリックして [Datasets] をクリックします。[Azure Blob Storage] > [Delimited Text] を選んだ後、入力データが格納されているパスなどを指定して Dataset を作成します。作成後の状態がこちらです。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/284925/43c9df4a-591f-2295-9aba-eacd442cb86c.png)

## 3-3. Datasets - ADLS
(Name: OrdersParquetByDate)
Parquet 出力データ用の Dataset です。[Data] ハブの [+] アイコンをクリックして [Datasets] をクリックします。[Azure Data Lake Storage Gen2] > [Parquet] を選んだ後、出力先のフォルダパスを指定して Dataset を作成します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/284925/a1c06572-bd1b-130d-2e06-bcdeb1e04d1f.png)
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/284925/5e799310-8fe5-f78c-6e2f-ca4ac1e04a2c.png)

これで Datasets の設定は完了です。Delta 用の Dataset については後続の手順で設定していきます。

# 4. Data flows 作成
CSV > Parquet 変換用の Data flow と、CSV > Delta 変換用の Data flow を作成していきます。

## 4-1. CSV > Parquet 変換用 Data flow
(Name: DataflowOrdersCsvToParquet)
[Develop] ハブの [+] アイコンをクリックして [Data flow] をクリックします。Data flow ではそれぞれの箱を「変換」(transformation) と呼びます。以下、各変換のキャプチャを貼付します。

### 4-1-1. Source 変換
(Name: SourceOrdersCsv)
Dataset に OrdersCsv を指定します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/284925/2f143a6e-08ef-b06e-ccd3-9fa6b2ece02e.png)

各カラムの型はデフォルトでは String で解釈されるため、*order_number* と *quantity* の Type を Integer に変更します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/284925/5a067d52-0744-6497-370c-cbddc3c33602.png)

### 4-1-2. Derived column 変換
(Name: AddDateColumns)
パーティションに利用する日時カラムを追加します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/284925/83212dfc-7a6b-af3a-70f2-9bd863a3a9bb.png)

### 4-1-3. Sink 変換
(Name: SinkOrdersParquetByDate)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/284925/9994a861-8dcf-2a8c-cf4f-b0c58573dbd3.png)

ここでは出力先フォルダーを毎回の処理実行時にクリアするように設定しています。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/284925/e447284a-a6ac-ea03-7876-1660e562f5d5.png)

*pt_year/pt_month/pt_day* というパーティションを指定します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/284925/5cb4c2ab-1ea0-22e0-e0cc-bf64978d32c5.png)

以下のカラムが出力ファイルに書き込まれます。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/284925/c7d9f90a-c576-690e-ecbe-89679d404a4a.png)

## 4-2. CSV > Delta 変換用 Data flow
(Name: DataflowOrdersCsvToDelta)

### 4-2-1. Source 変換
(Name: SourceOrdersCsv)
CSV > Parquet 変換用 Data flow と同じ内容のため割愛

### 4-2-2. Derived column 変換
(Name: AddDateColumns)
CSV > Parquet 変換用 Data flow と同じ内容のため割愛

### 4-2-3. Sink 変換
(Name: SinkOrdersDeltaByDate)

Delta を指定する場合、[Sink type] から [Delta] を指定します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/284925/5aeb903e-c33d-0ab1-b00e-873fcd16050c.png)

出力先 ADLS の Linked service を指定します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/284925/a188ef04-7061-1436-335e-a86eba5189d4.png)

[Settings] タブでフォルダーパス以外にも諸々の設定ができますが、今回は特に変更せずにデフォルトのままにしておきます。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/284925/dcb6267c-583b-3dcc-410c-375a116e8b8a.png)

*pt_year/pt_month/pt_day* というパーティションを指定します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/284925/e8c35fd8-8ae2-3469-3c48-f8c199c46ed6.png)

# 5. Pipelines 作成 & 実行
先ほど作成した Data flow を呼び出す Pipeline を作成し、デバッグ実行して結果を出力します。

## 5-1. CSV > Parquet 変換用 Data flow 用 Pipeline
(Name: PipelineDataflowOrdersCsvToParquet)

[Orchestrate] ハブの [+] アイコンをクリックして [Pipeline] をクリックします。[Move & transform] > [Data flow] を右側の空白部分にドラッグ & ドロップして先ほど作成したCSV > Parquet 変換用 Data flow を指定します。[Publish all] をクリックします。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/284925/c07f6e32-5b18-8d5c-c101-5ddecdf7c946.png)

[Add trigger] > [Trigger now] をクリックし、Pipeline を実行します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/284925/ef425dd9-2cec-5f5b-8349-928c2be867f0.png)

## 5-2. CSV > Delta 変換用 Data flow 用 Pipeline
(Name: PipelineDataflowOrdersCsvToDelta)
上記と同様の手順のため割愛します。

## 5-3. Pipeline 実行結果確認
[Monitor] ハブ > [Pipeline runs] から Pipeline の実行結果が確認できます。[STATUS] が [Succeeded] になっていれば OK です。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/284925/72c9acba-4588-7335-f3e4-91a455f9faac.png)

### Parquet 出力結果
実際にどのようなデータが出力されたかを見てみましょう。
[Data] ハブ > [Linked] の Storage accounts に ADLS のコンテナー一覧が表示されているはずなので、その中で出力先として利用しているコンテナーを指定します。
Parquet の方はこのような出力になるはずです。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/284925/1e3ee355-dcb8-2b5a-5ff0-bb56bef192b4.png)

`pt_year=2020` フォルダー (年のパーティション) と `_SUCCESS` ファイル (処理成功を表す) がある
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/284925/debeb6fa-230f-94d7-0ba8-ecfb46698df4.png)

`pt_month=07` フォルダー (月のパーティション) がある
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/284925/27629d2f-7997-921c-117b-54c4c08a3e23.png)

`pt_day=01` ~ `pt_day=07` フォルダー (日のパーティション) がある
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/284925/7f95ddb5-4c18-3879-d806-c77f7eb32ca4.png)

各フォルダーの配下に `part-00000-${GUID}.c0000.snappy.parquet` といったネーミングのファイルがある (これがデータの実体)
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/284925/0bb6be82-8aa5-e895-2363-59b501c817b3.png)

### Delta 出力結果
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/284925/de3e3978-1c05-6cc9-545f-3cea806b0d1f.png)

`pt_year=2020` フォルダー以下の構造は Parquet と同じですが一点だけ違いとして `_delta_log` フォルダーがある
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/284925/de017cb5-6349-3024-f39c-a67a844b1957.png)

`_delta_log` フォルダーには以下のような JSON ファイルがある
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/284925/1c8e1d75-e78c-4785-5ff6-2d6f92bd38c0.png)

JSON の中には以下のようなメタデータが格納されている。

```
{"commitInfo":{"timestamp":1595757668111,"operation":"WRITE","operationParameters":{"mode":"Append","partitionBy":"[\"pt_year\",\"pt_month\",\"pt_day\"]"},"isBlindAppend":true,"operationMetrics":{"numFiles":"7","numOutputBytes":"84256","numOutputRows":"1000"}}}
{"protocol":{"minReaderVersion":1,"minWriterVersion":2}}
{"metaData":{"id":"04e98468-c8ca-41e7-a01f-f589ab5f533b","format":{"provider":"parquet","options":{}},"schemaString":"{\"type\":\"struct\",\"fields\":[{\"name\":\"order_number\",\"type\":\"integer\",\"nullable\":true,\"metadata\":{}},{\"name\":\"created_at\",\"type\":\"string\",\"nullable\":true,\"metadata\":{}},{\"name\":\"customer_id\",\"type\":\"string\",\"nullable\":true,\"metadata\":{}},{\"name\":\"stock_name\",\"type\":\"string\",\"nullable\":true,\"metadata\":{}},{\"name\":\"quantity\",\"type\":\"integer\",\"nullable\":true,\"metadata\":{}},{\"name\":\"pt_year\",\"type\":\"string\",\"nullable\":true,\"metadata\":{}},{\"name\":\"pt_month\",\"type\":\"string\",\"nullable\":true,\"metadata\":{}},{\"name\":\"pt_day\",\"type\":\"string\",\"nullable\":true,\"metadata\":{}}]}","partitionColumns":["pt_year","pt_month","pt_day"],"configuration":{},"createdTime":1595757649086}}
{"add":{"path":"pt_year=2020/pt_month=07/pt_day=01/part-00000-4dfe08ac-252a-4cee-a797-e7bdd4b48fa5.c000.snappy.parquet","partitionValues":{"pt_year":"2020","pt_month":"07","pt_day":"01"},"size":14098,"modificationTime":1595757657000,"dataChange":true}}
{"add":{"path":"pt_year=2020/pt_month=07/pt_day=02/part-00000-e12e27d8-e651-4bd2-8e45-6dddac0665ac.c000.snappy.parquet","partitionValues":{"pt_year":"2020","pt_month":"07","pt_day":"02"},"size":12294,"modificationTime":1595757658000,"dataChange":true}}
{"add":{"path":"pt_year=2020/pt_month=07/pt_day=03/part-00000-bb8e1936-e220-4200-a241-57d1663b00e4.c000.snappy.parquet","partitionValues":{"pt_year":"2020","pt_month":"07","pt_day":"03"},"size":11243,"modificationTime":1595757660000,"dataChange":true}}
{"add":{"path":"pt_year=2020/pt_month=07/pt_day=04/part-00000-5968555c-de58-4c29-b1b0-d433870d7976.c000.snappy.parquet","partitionValues":{"pt_year":"2020","pt_month":"07","pt_day":"04"},"size":11288,"modificationTime":1595757661000,"dataChange":true}}
{"add":{"path":"pt_year=2020/pt_month=07/pt_day=05/part-00000-b19c8e6b-1853-46c6-9d34-5b8b70f50d8f.c000.snappy.parquet","partitionValues":{"pt_year":"2020","pt_month":"07","pt_day":"05"},"size":11764,"modificationTime":1595757662000,"dataChange":true}}
{"add":{"path":"pt_year=2020/pt_month=07/pt_day=06/part-00000-725b1fcd-5c6c-49bf-a194-56710d5766fc.c000.snappy.parquet","partitionValues":{"pt_year":"2020","pt_month":"07","pt_day":"06"},"size":11898,"modificationTime":1595757664000,"dataChange":true}}
{"add":{"path":"pt_year=2020/pt_month=07/pt_day=07/part-00000-27b18250-ae7e-42de-8f76-37c69fac7d07.c000.snappy.parquet","partitionValues":{"pt_year":"2020","pt_month":"07","pt_day":"07"},"size":11671,"modificationTime":1595757665000,"dataChange":true}}
```

# 6. SQL オンデマンドでのクエリ実行
以下、Delta 形式のデータに対するクエリの例をいくつか記載します。Parquet 形式についても同様のクエリでアクセスできます。他のパターンについては以下ドキュメントを参考頂くのが良いと思います。

[データ アナリスト チュートリアル: SQL オンデマンド (プレビュー) を使用して Azure Synapse Studio (プレビュー) で Azure Open Datasets を分析する - Azure Synapse Analytics | Microsoft Docs](https://docs.microsoft.com/ja-jp/azure/synapse-analytics/sql/tutorial-data-analyst)

## 単一ファイル指定 & TOP 100 
```
SELECT
    TOP 100 *
FROM
    OPENROWSET(
        BULK 'https://stnakazaxqiitadf.dfs.core.windows.net/orders-container/orders_delta_by_date/pt_year=2020/pt_month=07/pt_day=01/part-00000-4dfe08ac-252a-4cee-a797-e7bdd4b48fa5.c000.snappy.parquet',
        FORMAT='PARQUET'
    ) AS [r];
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/284925/d5e6a5cc-f95a-0aaa-40b8-d208c4221399.png)

## 単一ファイル指定 & COUNT
```
SELECT
    COUNT(*)
FROM
    OPENROWSET(
        BULK 'https://stnakazaxqiitadf.dfs.core.windows.net/orders-container/orders_delta_by_date/pt_year=2020/pt_month=07/pt_day=01/part-00000-4dfe08ac-252a-4cee-a797-e7bdd4b48fa5.c000.snappy.parquet',
        FORMAT='PARQUET'
    ) AS [r];
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/284925/dfa1f6d6-81dd-9022-15c2-13d45abacf8f.png)

## 全ファイル指定 & COUNT
```
SELECT
    COUNT(*)
FROM
    OPENROWSET(
        BULK 'https://stnakazaxqiitadf.dfs.core.windows.net/orders-container/orders_delta_by_date/pt_year=*/pt_month=*/pt_day=*/*.parquet',
        FORMAT='PARQUET'
    ) AS [r];
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/284925/0170e299-1800-65a8-42be-1ddee4779bfc.png)

## WHERE 句で絞り込んだ日付別の件数
```
SELECT
    SUBSTRING(r.created_at, 0, 11) AS created_at_date,
    COUNT(*) AS count
FROM
    OPENROWSET(
        BULK 'https://stnakazaxqiitadf.dfs.core.windows.net/orders-container/orders_delta_by_date/pt_year=2020/pt_month=07/pt_day=*/*.parquet',
        FORMAT='PARQUET'
    ) AS [r]
WHERE r.filepath(1) >= '01' AND r.filepath(1) <= '04'
GROUP BY SUBSTRING(r.created_at, 0, 11)
ORDER BY created_at_date ASC;
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/284925/b291e380-6478-9aea-693f-e2c66379370a.png)

以上です。