【Azure Databricks】CSVファイルの読込とテーブルへのマージ

1.はじめに
2.前提要件
3.CSVファイルをテーブルに格納する
- 3-1.ストレージアカウントへcustomer.csv ファイルをインポートする
- 3-2.ノートブックを使用して、CSVファイルをテーブルに格納する
4.まとめ
- この記事を読んだ方へのオススメコンテンツはこちら

1.はじめに

皆さんこんにちは。

今回は、CSVファイルを取込む方法について説明していきます。

2.前提要件

本書を実施する際の前提条件は

ストレージアカウントが作成済みであること
Access connectorが作成済みであること
ストレージアカウントのBLOB データ共同作成者ロールをAccess connectorに付与していること
Metastoreが作成済みであること
Unity CatalogのDatabricks ワークスペースを有効にしていること
ストレージの認証情報を作成済みであること
Unity Catalog 外部ロケーションを作成済みであること
操作ユーザはUnity Catalogのテーブル更新権限があること

以上が作成済み、指定済みであることを前提としています。

3.CSVファイルをテーブルに格納する

3-1.ストレージアカウントへcustomer.csv ファイルをインポートする

① ストレージアカウントの画面でコンテナーを作成し、CSVファイルを格納します。

② サイドバーで「コンテナー」を選択し、「コンテナー」をクリックします。

③ コンテナーに「container-name」等の名前を付けます。

④「作成」ボタンをクリックします。

⑤ コンテナーの一覧が表示されます。

作成したコンテナーをクリックします。

⑥ 次に、フォルダーを作成し、CSVファイルを格納します。

⑦「ディレクトリーの追加」をクリックします。

⑧ フォルダーに「raw」等の名前を付けます。

⑨「保存」ボタンをクリックします。

⑩ フォルダーの一覧が表示されます。

作成したフォルダーをクリックします。

⑪ 作成したフォルダーにて「アップロード」をクリックし、「ファイルを参照」をクリックします。

⑫ ダイアログでローカルコンピューターのファイルをアップロードできます。customer.csv ファイルを選択してください。

⑬ ファイルを選択して、「アップロード」ボタンをクリックします。

ストレージアカウントの「raw」フォルダーにcustomer.csv ファイルがアップロードされました。

3-2.ノートブックを使用して、CSVファイルをテーブルに格納する

ワークスペースでノートブックを作成してCSVファイルを読み込みます。次に、 Unity Catalog でテーブルを作成して、CSVファイルをマージします。

① Databricksワークスペースをログインします。

② Databricks画面でサイドバーから「 Data science & Engineering 」をクリックします。

③サイドバーから「新規」をクリックして「ノートブック」を選択します。

④ダイアログの「名前」項目でノートブック名を入力します。（例： convert-notebook ）

⑤「デフォルト言語」項目で「Python」を選択します。

⑥「クラスター」項目で存在するクラスターを選択します。

⑦「作成」ボタンをクリックすると、ノートブックが表示されます。

⑧ 次に、CSVファイルを読み取って、「demotempview」一時的なビューにデータを入力します。

以下のコードをノートブックの2行目にコピーして、「Shift」+「Enter」キーを押します。

※ 文法

CREATE [OR REPLACE] [TEMPORARY] VIEW [IF NOT EXISTS] view_name
USING format 
OPTIONS (option1 "value1", option2 "value2", ...)

CREATE [OR REPLACE] [TEMPORARY] VIEW [IF NOT EXISTS] view_name

USING format

OPTIONS (option1 "value1", option2 "value2", ...)

※ 例

CREATE or replace TEMPORARY VIEW demotempview
USING CSV
OPTIONS (path "abfss://<your_container_name>@<your_storage_account_name>.dfs.core.windows.net/<your_folder_name>/", header "true", mode "FAILFAST")

CREATE or replace TEMPORARY VIEW demotempview

USING CSV

OPTIONS (path "abfss://<your_container_name>@<your_storage_account_name>.dfs.core.windows.net/<your_folder_name>/", header "true", mode "FAILFAST")

⑨ データをビューに格納したかを確認できるように、以下のコマンドを３行目にコピーして「Shift」+「Enter」キーを押します。

実行後、結果が画像のように表示されます。

CSVファイルを読み取ることができます。

※ 文法

SELECT * FROM view_name

1	SELECT * FROM view_name

※ 例

select * from demotempview

1	select * from demotempview

⑩ 次に、 Unity Catalog でカタログ、スキーマとテーブルを作成して、「demotempview」ビューのデータをマージします。

以下のコマンドを4行目にコピーして「Shift」+「Enter」キーを押します。「MyCatalog」カタログ、「MySchema」スキーマ及び、「MyTable」テーブルが作成されます。

※ 文法

CREATE CATALOG [IF NOT EXISTS] catalog_name;
CREATE SCHEMA [IF NOT EXISTS] catalog_name.schema_name;
CREATE TABLE [IF NOT EXISTS] catalog_name.schema_name.table_name ( column_specification )

CREATE CATALOG [IF NOT EXISTS] catalog_name;

CREATE SCHEMA [IF NOT EXISTS] catalog_name.schema_name;

CREATE TABLE [IF NOT EXISTS] catalog_name.schema_name.table_name ( column_specification )

※ 例

CREATE CATALOG IF NOT EXISTS MyCatalog;
CREATE SCHEMA IF NOT EXISTS MyCatalog.MySchema;
CREATE TABLE IF NOT EXISTS MyCatalog.MySchema.MyTable(
	customer_id bigint,
	tax_id int,
	tax_code int,
	customer_name string,
	state string,
	city string,
	postcode float,
	street string
)

CREATE CATALOG IF NOT EXISTS MyCatalog;

CREATE SCHEMA IF NOT EXISTS MyCatalog.MySchema;

CREATE TABLE IF NOT EXISTS MyCatalog.MySchema.MyTable(

customer_id bigint,

tax_id int,

tax_code int,

customer_name string,

state string,

city string,

postcode float,

street string

)

⑪ テーブルの作成後、以下のコマンドを実行して、「demotempview」ビューのデータをテーブルにマージします。

以下のコマンドを5行目にコピーして「Shift」+「Enter」キーを押します。

実行後、画像のように9819行がテーブルにマージされました。

※ 文法

MERGE INTO target_table_name [target_alias]
USING source_table_reference [source_alias]
ON merge_condition
{ WHEN MATCHED [ AND matched_condition ] THEN matched_action |
WHEN NOT MATCHED [BY TARGET] [ AND not_matched_condition ] THEN not_matched_action |
WHEN NOT MATCHED BY SOURCE [ AND not_matched_by_source_condition ] THEN not_matched_by_source_action } [...]
matched_action
{ DELETE |
UPDATE SET * |
UPDATE SET { column = { expr | DEFAULT } } [, ...] }
not_matched_action
{ INSERT * |
INSERT (column1 [, ...] ) VALUES ( expr | DEFAULT ] [, ...] )
not_matched_by_source_action
{ DELETE |
UPDATE SET { column = { expr | DEFAULT } } [, ...] }

MERGE INTO target_table_name [target_alias]

USING source_table_reference [source_alias]

ON merge_condition

{ WHEN MATCHED [ AND matched_condition ] THEN matched_action |

WHEN NOT MATCHED [BY TARGET] [ AND not_matched_condition ] THEN not_matched_action |

WHEN NOT MATCHED BY SOURCE [ AND not_matched_by_source_condition ] THEN not_matched_by_source_action } [...]

matched_action

{ DELETE |

UPDATE SET * |

UPDATE SET { column = { expr | DEFAULT } } [, ...] }

not_matched_action

{ INSERT * |

INSERT (column1 [, ...] ) VALUES ( expr | DEFAULT ] [, ...] )

not_matched_by_source_action

{ DELETE |

UPDATE SET { column = { expr | DEFAULT } } [, ...] }

※ 例

MERGE INTO MyCatalog.MySchema.MyTable a
USING demotempview b
ON a.customer_id = b.customer_id
WHEN MATCHED THEN
		UPDATE SET customer_id = b.customer_id,
				tax_id = b.tax_id,
				tax_code = b.tax_code,
				customer_name = b.customer_name,
				state = b.state,
				city = b.city,
				postcode = b.postcode,
				street = b.street                
WHEN NOT MATCHED THEN
		INSERT (customer_id,tax_id,tax_code,
				customer_name,state,city,
				postcode,street)
		VALUES (b.customer_id,b.tax_id,b.tax_code,
				b.customer_name,b.state,b.city,
				b.postcode,b.street)

MERGE INTO MyCatalog.MySchema.MyTable a

USING demotempview b

ON a.customer_id = b.customer_id

WHEN MATCHED THEN

UPDATE SET customer_id = b.customer_id,

tax_id = b.tax_id,

tax_code = b.tax_code,

customer_name = b.customer_name,

state = b.state,

city = b.city,

postcode = b.postcode,

street = b.street

WHEN NOT MATCHED THEN

INSERT (customer_id,tax_id,tax_code,

customer_name,state,city,

postcode,street)

VALUES (b.customer_id,b.tax_id,b.tax_code,

b.customer_name,b.state,b.city,

b.postcode,b.street)

⑫ データをテーブルにマージしたかを確認できるように以下のコマンドを実行します。

以下のコマンドを6行目にコピーして「Shift」+「Enter」キーを押します。

実行後、結果が画像のように表示されます。

※ 文法

SELECT * FROM catalog_name.schema_name.table_name;

1	SELECT * FROM catalog_name.schema_name.table_name;

※ 例

SELECT * FROM MyCatalog.MySchema.MyTable;

1	SELECT * FROM MyCatalog.MySchema.MyTable;

4.まとめ

これでCSVファイルを取込む方法について説明しました。

今回の記事が少しでもDatabricksを知るきっかけや、業務のご参考になれば幸いです。

日商エレクトロニクスでは、Azure Databricksの環境構築パッケージを用意しています。

Azure DatabricksやAzure活用、マイクロソフト製品の活用についてご相談事がありましたらぜひお問い合わせください！

・Azure Databricks連載シリーズはこちら

お問い合わせはこちら

この記事を読んだ方へのオススメコンテンツはこちら

アセスメントサービス

この記事を書いた人

phongcq

この投稿者の最新の記事

この投稿者の記事一覧

2024年11月6日ブログ【Azure Databricks】Lakehouse フェデレーションと外部カタログについての説明
2024年5月15日ブログ【Microsoft Fabric】Synapse Real-Time Analytics ユーザー向け Fabric チュートリアル
2024年4月5日ブログ【Microsoft Fabric】Data Factory ユーザー向け Fabric チュートリアル
2024年3月18日ブログ【Microsoft Fabric】Power BI ユーザー向け Fabric チュートリアル

【Azure Databricks】CSVファイルの読込とテーブルへのマージ

1.はじめに

2.前提要件

3.CSVファイルをテーブルに格納する

3-1.ストレージアカウントへcustomer.csv ファイルをインポートする

3-2.ノートブックを使用して、CSVファイルをテーブルに格納する

4.まとめ

この記事を読んだ方へのオススメコンテンツはこちら

この記事を書いた人

この投稿者の最新の記事

Add comment コメントをキャンセル

03-6272-5011