Azure Blob Storage を外部ステージに設定し Snowpipe によるデータロードを試してみる #SnowflakeDB

			
			az provider register --namespace Microsoft.EventGrid
$ az provider show --namespace Microsoft.EventGrid --query "registrationState"
"Registered"

		

			
			# 環境変数を設定
resource_group_name="rg-test" # 作成済みのリソースグループ名
data_storage_account_name="<Blobストレージの作成先ストレージアカウント名>"
storage_queue_account_name="<ストレージキューの作成先となるストレージアカウント名>"
queue_name="snowpipe-queue" # キューの名前
subscription_name="snowpipe-subscription" # サブスクリプション名

# ストレージキュー用ストレージアカウントを作成
az storage account create \
  --resource-group $resource_group_name \
  --name $storage_queue_account_name \
  --location japaneast \
  --sku Standard_LRS \
  --kind StorageV2 

# ストレージキューを作成
az storage queue create \
  --name $queue_name \
  --account-name $storage_queue_account_name

		

			
			$ az extension add --name eventgrid
No stable version of 'eventgrid' to install. Preview versions allowed.
The installed extension 'eventgrid' is in preview.

		

			
			# データ用ストレージアカウントIDを取得 (Event Gridのソース)
storageid=$(az storage account show \
  --name $data_storage_account_name \
  --resource-group $resource_group_name \
  --query id \
  --output tsv)

# キュー用ストレージアカウントIDを取得 (Event Gridの宛先)
queue_storage_id=$(az storage account show \
  --name $storage_queue_account_name \
  --resource-group $resource_group_name \
  --query id \
  --output tsv)

# キュー用ストレージアカウントIDを使用し、キューのIDを作成
queueid="$queue_storage_id/queueservices/default/queues/$queue_name"

# Event Gridサブスクリプションを作成
az eventgrid event-subscription create \
  --source-resource-id $storageid \
  --name $subscription_name \
  --endpoint-type storagequeue \
  --endpoint $queueid \
  --advanced-filter data.api stringin CopyBlob PutBlob PutBlockList FlushWithClose SftpCommit

		

			
			--通知統合を作成
USE ROLE ACCOUNTADMIN;

CREATE NOTIFICATION INTEGRATION pipe_queue_int
  ENABLED = true
  TYPE = QUEUE
  NOTIFICATION_PROVIDER = AZURE_STORAGE_QUEUE
  AZURE_STORAGE_QUEUE_PRIMARY_URI = '<ストレージキューURL>'
  AZURE_TENANT_ID = '<テナントID>';

		

			
			DESC NOTIFICATION INTEGRATION pipe_queue_int;

		

			
			CREATE STAGE my_azure_stage
  STORAGE_INTEGRATION = my_azure_storage_int
  URL = 'azure://<ストレージアカウント名>.blob.core.windows.net/<コンテナー名>/';

		

			
			--ファイルフォーマットを作成
CREATE OR REPLACE FILE FORMAT my_csv_format
  TYPE = CSV
  FIELD_DELIMITER = ','
  PARSE_HEADER = TRUE
  EMPTY_FIELD_AS_NULL = true
  COMPRESSION = AUTO;

--スキーマ検出機能によりテーブルを定義
>CREATE OR REPLACE TABLE mytable
    USING TEMPLATE (
        SELECT ARRAY_AGG(OBJECT_CONSTRUCT(*))
        FROM TABLE(
            INFER_SCHEMA(
                LOCATION=>'@my_azure_stage/',
                FILE_FORMAT=>'my_csv_format'
                )
            )
        );
+-------------------------------------+
| status                              |
|-------------------------------------|
| Table MYTABLE successfully created. |
+-------------------------------------+

		

			
			CREATE OR REPLACE PIPE mypipe
AUTO_INGEST = true
INTEGRATION = 'PIPE_QUEUE_INT'
AS
COPY INTO test_db.public.mytable
  FROM @test_db.public.my_azure_stage/
  FILE_FORMAT=(FORMAT_NAME = 'my_csv_format')
	MATCH_BY_COLUMN_NAME = CASE_INSENSITIVE;

		

			
			>SELECT * FROM MYTABLE;
+----+-----+----------+------+
| ID | Val | Category | Date |
|----+-----+----------+------|
+----+-----+----------+------+

		

			
			>SELECT * FROM MYTABLE;
+----+--------------+----------+-----------+
| ID |          Val | Category | Date      |
|----+--------------+----------+-----------|
|  1 | -0.359878994 | A        | 2024/4/13 |
|  2 | -2.681319922 | B        | 2024/4/13 |
|  3 | -0.929509687 | B        | 2024/4/13 |
|  4 | -0.682002319 | A        | 2024/4/13 |
|  5 | -1.241710994 | C        | 2024/4/13 |
|  6 |  0.769089496 | B        | 2024/4/13 |
|  7 | -0.151905356 | B        | 2024/4/13 |
|  8 |  0.462443169 | C        | 2024/4/13 |
|  9 | -0.453144568 | A        | 2024/4/13 |
| 10 |  0.582974854 | B        | 2024/4/13 |
+----+--------------+----------+-----------+

		

			
			SELECT * FROM TABLE(information_schema.copy_history(
                        table_name => 'mytable',
                        start_time => DATEADD(hour, -1, current_timestamp())
                        ));
+--------------------------+----------------------------------------------------------------------+-------------------------------+-----------+------------+-----------+---------------------+-------------------------+---------------------------+-------------------------+-------------+-------------+--------+--------------------+-------------------+------------+-------------------+------------------+-----------+-------------------------------+--------------+
| FILE_NAME                | STAGE_LOCATION                                                       | LAST_LOAD_TIME                | ROW_COUNT | ROW_PARSED | FILE_SIZE | FIRST_ERROR_MESSAGE | FIRST_ERROR_LINE_NUMBER | FIRST_ERROR_CHARACTER_POS | FIRST_ERROR_COLUMN_NAME | ERROR_COUNT | ERROR_LIMIT | STATUS | TABLE_CATALOG_NAME | TABLE_SCHEMA_NAME | TABLE_NAME | PIPE_CATALOG_NAME | PIPE_SCHEMA_NAME | PIPE_NAME | PIPE_RECEIVED_TIME            | BYTES_BILLED |
|--------------------------+----------------------------------------------------------------------+-------------------------------+-----------+------------+-----------+---------------------+-------------------------+---------------------------+-------------------------+-------------+-------------+--------+--------------------+-------------------+------------+-------------------+------------------+-----------+-------------------------------+--------------|
| sample_data_20240413.csv | azure://<ストレージアカウント>.blob.core.windows.net/<コンテナ名>/      | 2025-10-10 00:42:05.182 -0700 |        10 |         10 |       300 | NULL                |                    NULL |                      NULL | NULL                    |           0 |           1 | Loaded | TEST_DB            | PUBLIC            | MYTABLE    | TEST_DB           | PUBLIC           | MYPIPE    | 2025-10-10 00:41:44.725 -0700 |          300 |
+--------------------------+----------------------------------------------------------------------+-------------------------------+-----------+------------+-----------+---------------------+-------------------------+---------------------------+-------------------------+-------------+-------------+--------+--------------------+-------------------+------------+-------------------+------------------+-----------+-------------------------------+--------------+

		

			
			>ALTER PIPE mypipe REFRESH;
+--------------------------+--------+
| File                     | Status |
|--------------------------+--------|
| sample_data_20240412.csv | SENT   |
| sample_data_20240413.csv | SENT   |
+--------------------------+--------+

>SELECT * FROM MYTABLE;
+----+--------------+----------+-----------+
| ID |          Val | Category | Date      |
|----+--------------+----------+-----------|
|  1 |  1.114388475 | A        | 2024/4/12 |
|  2 | -0.617730236 | B        | 2024/4/12 |
|  3 | -1.226039849 | A        | 2024/4/12 |
|  4 | -1.249994022 | A        | 2024/4/12 |
|  5 |  0.775035178 | A        | 2024/4/12 |
|  6 |  0.442294860 | C        | 2024/4/12 |
|  7 |  0.819239543 | A        | 2024/4/12 |
|  8 | -0.921003054 | B        | 2024/4/12 |
|  9 | -0.288820276 | B        | 2024/4/12 |
| 10 |  0.752006217 | B        | 2024/4/12 |
|  1 | -0.359878994 | A        | 2024/4/13 |
|  2 | -2.681319922 | B        | 2024/4/13 |
|  3 | -0.929509687 | B        | 2024/4/13 |
|  4 | -0.682002319 | A        | 2024/4/13 |
|  5 | -1.241710994 | C        | 2024/4/13 |
|  6 |  0.769089496 | B        | 2024/4/13 |
|  7 | -0.151905356 | B        | 2024/4/13 |
|  8 |  0.462443169 | C        | 2024/4/13 |
|  9 | -0.453144568 | A        | 2024/4/13 |
| 10 |  0.582974854 | B        | 2024/4/13 |
+----+--------------+----------+-----------+

		

			
			>SELECT * FROM TABLE(information_schema.copy_history(
			                    table_name => 'mytable',
			                    start_time => DATEADD(hour, -1, current_timestamp())
			                    ));
+--------------------------+----------------------------------------------------------------------+-------------------------------+-----------+------------+-----------+---------------------+-------------------------+---------------------------+-------------------------+-------------+-------------+--------+--------------------+-------------------+------------+-------------------+------------------+-----------+-------------------------------+--------------+
| FILE_NAME                | STAGE_LOCATION                                                       | LAST_LOAD_TIME                | ROW_COUNT | ROW_PARSED | FILE_SIZE | FIRST_ERROR_MESSAGE | FIRST_ERROR_LINE_NUMBER | FIRST_ERROR_CHARACTER_POS | FIRST_ERROR_COLUMN_NAME | ERROR_COUNT | ERROR_LIMIT | STATUS | TABLE_CATALOG_NAME | TABLE_SCHEMA_NAME | TABLE_NAME | PIPE_CATALOG_NAME | PIPE_SCHEMA_NAME | PIPE_NAME | PIPE_RECEIVED_TIME            | BYTES_BILLED |
|--------------------------+----------------------------------------------------------------------+-------------------------------+-----------+------------+-----------+---------------------+-------------------------+---------------------------+-------------------------+-------------+-------------+--------+--------------------+-------------------+------------+-------------------+------------------+-----------+-------------------------------+--------------|
| sample_data_20240413.csv | azure://<ストレージアカウント>.blob.core.windows.net/<コンテナ名>/      | 2025-10-10 00:42:05.182 -0700 |        10 |         10 |       300 | NULL                |                    NULL |                      NULL | NULL                    |           0 |           1 | Loaded | TEST_DB            | PUBLIC            | MYTABLE    | TEST_DB           | PUBLIC           | MYPIPE    | 2025-10-10 00:41:44.725 -0700 |          300 |
| sample_data_20240412.csv | azure://<ストレージアカウント>.blob.core.windows.net/<コンテナ名>/      | 2025-10-10 00:45:53.183 -0700 |        10 |         10 |       297 | NULL                |                    NULL |                      NULL | NULL                    |           0 |           1 | Loaded | TEST_DB            | PUBLIC            | MYTABLE    | TEST_DB           | PUBLIC           | MYPIPE    | 2025-10-10 00:45:27.489 -0700 |          297 |
+--------------------------+----------------------------------------------------------------------+-------------------------------+-----------+------------+-----------+---------------------+-------------------------+---------------------------+-------------------------+-------------+-------------+--------+--------------------+-------------------+------------+-------------------+------------------+-----------+-------------------------------+--------------+

		

			
			--「path1/」配下のデータファイルをロードするためのテーブルとパイプを定義
CREATE OR REPLACE TABLE mytable_path1
    USING TEMPLATE (
        SELECT ARRAY_AGG(OBJECT_CONSTRUCT(*))
        FROM TABLE(
            INFER_SCHEMA(
                LOCATION=>'@my_azure_stage/',
                FILE_FORMAT=>'my_csv_format'
                )
            )
        );

-- ステージの「path1/」配下のデータファイルを「mytable_path1」にSnowpipeでロード
CREATE OR REPLACE PIPE mypipe_path1
AUTO_INGEST = true
INTEGRATION = 'PIPE_QUEUE_INT'
AS
COPY INTO test_db.public.mytable_path1
  FROM @test_db.public.my_azure_stage/path1/
  FILE_FORMAT=(FORMAT_NAME = 'my_csv_format')
	MATCH_BY_COLUMN_NAME = CASE_INSENSITIVE;

SELECT * FROM mytable_path1;

--「path2/」配下のデータファイルをロードするためのテーブルとパイプを定義
CREATE OR REPLACE TABLE mytable_path2
    USING TEMPLATE (
        SELECT ARRAY_AGG(OBJECT_CONSTRUCT(*))
        FROM TABLE(
            INFER_SCHEMA(
                LOCATION=>'@my_azure_stage/',
                FILE_FORMAT=>'my_csv_format'
                )
            )
        );

-- ステージの「path2/」配下のデータファイルを「mytable_path2」にSnowpipeでロード
CREATE OR REPLACE PIPE mypipe_path2
AUTO_INGEST = true
INTEGRATION = 'PIPE_QUEUE_INT'
AS
COPY INTO test_db.public.mytable_path2
  FROM @test_db.public.my_azure_stage/path2/
  FILE_FORMAT=(FORMAT_NAME = 'my_csv_format')
	MATCH_BY_COLUMN_NAME = CASE_INSENSITIVE;

SELECT * FROM mytable_path2;

		

			
			>SELECT COUNT(*) FROM mytable_path1;
+----------+
| COUNT(*) |
|----------|
|       10 |
+----------+

>SELECT * FROM mytable_path1 ORDER BY 2 LIMIT 1;
+----+-----------+----------+--------------+
| ID | Date      | Category |          Val |
|----+-----------+----------+--------------|
|  1 | 2024/4/14 | C        | -0.939345586 |
+----+-----------+----------+--------------+

		

			
			>SELECT COUNT(*) FROM mytable_path2;
+----------+
| COUNT(*) |
|----------|
|       10 |
+----------+

>SELECT * FROM mytable_path2 ORDER BY 2 LIMIT 1;
+-----------+----+-------------+----------+
| Date      | ID |         Val | Category |
|-----------+----+-------------+----------|
| 2024/4/15 |  1 | 0.331979236 | B        |
+-----------+----+-------------+----------+

		

			
			ALTER STORAGE INTEGRATION my_azure_storage_int SET
   STORAGE_ALLOWED_LOCATIONS = (
	   'azure://<ストレージアカウント名>.blob.core.windows.net/test-container/',
	   'azure://<ストレージアカウント名>.blob.core.windows.net/test-container2/');

		

			
			CREATE STAGE my_azure_stage2
  STORAGE_INTEGRATION = my_azure_storage_int
  URL = 'azure://<ストレージアカウント名>.blob.core.windows.net/test-container2/';

		

			
			--テーブルとパイプを作成
CREATE OR REPLACE TABLE mytable_container2
    USING TEMPLATE (
        SELECT ARRAY_AGG(OBJECT_CONSTRUCT(*))
        FROM TABLE(
            INFER_SCHEMA(
                LOCATION=>'@my_azure_stage/',
                FILE_FORMAT=>'my_csv_format'
                )
            )
        );

CREATE OR REPLACE PIPE mypipe_container2
AUTO_INGEST = true
INTEGRATION = 'PIPE_QUEUE_INT'
AS
COPY INTO test_db.public.mytable_container2
  FROM @test_db.public.my_azure_stage2/
  FILE_FORMAT=(FORMAT_NAME = 'my_csv_format')
	MATCH_BY_COLUMN_NAME = CASE_INSENSITIVE;

		

			
			>SELECT COUNT(*) FROM mytable_container2;
+----------+
| COUNT(*) |
|----------|
|       10 |
+----------+

>SELECT * FROM mytable_container2 LIMIT 1;
+-----------+-------------+----------+----+
| Date      |         Val | Category | ID |
|-----------+-------------+----------+----|
| 2024/4/17 | 0.013401644 | C        |  1 |
+-----------+-------------+----------+----+

		

			
			# 環境変数を設定
resource_group_name="rg-test"
new_storage_account_name='<新規作成したストレージアカウント名>'
storage_queue_account_name="<ストレージキュー名>"
queue_name="<既存のキューの名前>"
subscription_name="<新規作成するサブスクリプション名>"

# データ用ストレージアカウントIDを取得 (Event Gridのソースとなる新規作成したストレージアカウント)
storageid=$(az storage account show \
  --name $new_storage_account_name \
  --resource-group $resource_group_name \
  --query id \
  --output tsv)

# キュー用ストレージアカウントIDを取得 (Event Gridの宛先)
queue_storage_id=$(az storage account show \
  --name $storage_queue_account_name \
  --resource-group $resource_group_name \
  --query id \
  --output tsv)

# キュー用ストレージアカウントIDを使用しキューのIDを作成
queueid="$queue_storage_id/queueservices/default/queues/$queue_name"

# Event Gridサブスクリプションを作成
az eventgrid event-subscription create \
  --source-resource-id $storageid \
  --name $subscription_name \
  --endpoint-type storagequeue \
  --endpoint $queueid \
  --advanced-filter data.api stringin CopyBlob PutBlob PutBlockList FlushWithClose SftpCommit

		

			
			ALTER STORAGE INTEGRATION my_azure_storage_int SET
   STORAGE_ALLOWED_LOCATIONS = (
	   'azure://<既存のストレージアカウント名>.blob.core.windows.net/test-container/',
	   'azure://<既存のストレージアカウント名>.blob.core.windows.net/test-container2/',
	   'azure://<新規作成したストレージアカウント名>.blob.core.windows.net/new-container/');

		

			
			--新しいコンテナ向けのステージを作成
CREATE STAGE my_azure_stage3
  STORAGE_INTEGRATION = my_azure_storage_int
  URL = 'azure://<新規作成したストレージアカウント名>.blob.core.windows.net/new-container/';

		

			
			--テーブルとパイプを作成
CREATE OR REPLACE TABLE mytable_storageaccount2
    USING TEMPLATE (
        SELECT ARRAY_AGG(OBJECT_CONSTRUCT(*))
        FROM TABLE(
            INFER_SCHEMA(
                LOCATION=>'@my_azure_stage/',
                FILE_FORMAT=>'my_csv_format'
                )
            )
        );

CREATE OR REPLACE PIPE mypipe_storageaccount2
AUTO_INGEST = true
INTEGRATION = 'PIPE_QUEUE_INT'
AS
COPY INTO test_db.public.mytable_storageaccount2
  FROM @test_db.public.my_azure_stage3/
  FILE_FORMAT=(FORMAT_NAME = 'my_csv_format')
	MATCH_BY_COLUMN_NAME = CASE_INSENSITIVE;

		

			
			>SELECT COUNT(*) FROM mytable_storageaccount2;
+----------+
| COUNT(*) |
|----------|
|       10 |
+----------+

>SELECT * FROM mytable_storageaccount2 ORDER BY 2 LIMIT 1;
+-----------+--------------+----+----------+
| Date      |          Val | ID | Category |
|-----------+--------------+----+----------|
| 2024/4/18 | -1.033691009 |  4 | A        |
+-----------+--------------+----+----------+