Glue の Visual ETL で Snowflake から Redshift Serverless に連携

# はじめに

AWS Glue の Visual ETL 機能を使用すると、コードをほとんど書かずにデータ連携パイプラインを構築できます。本記事では、Snowflake から Amazon Redshift Serverless へのデータ連携を Visual ETL で実装する方法を紹介します。

Snowflake の検証は、クレジットカード登録不要の無料トライアル (30日間) を活用しています。



# Snowflake の Free Trial アカウント作成

Snowflake の Free Trial アカウントを作成します。Enterprise Edition を選択します。

![image-20250420104848253.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1002774/673bac59-3c9e-4ee2-94fe-f0fd1d8263d3.png)



すると、すぐに Snowflake にログインができました。


![image-20250420105022677.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1002774/283b6c40-1ef8-4fde-9fae-4e02f5391498.png)



Projects - Worksheets から適当なサンプルデータを Snowflake に入れます。

![image-20250427114844000.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1002774/82ebd8ea-d24c-4bee-a94d-c719137eda29.png)



データを入れる : TEST_DATABASE

```
-- サンプルテーブルの作成
CREATE OR REPLACE TABLE customer_sample (
    customer_id INTEGER,
    first_name VARCHAR(50),
    last_name VARCHAR(50),
    email VARCHAR(100),
    age INTEGER,
    city VARCHAR(50),
    registration_date DATE,
    total_purchase DECIMAL(10,2)
);

-- サンプルデータの挿入
INSERT INTO customer_sample VALUES
    (1, '田中', '太郎', 'tanaka.taro@example.com', 35, '東京', '2022-01-15', 25000.50),
    (2, '佐藤', '花子', 'sato.hanako@example.com', 28, '大阪', '2022-02-20', 18750.75),
    (3, '鈴木', '一郎', 'suzuki.ichiro@example.com', 42, '名古屋', '2022-03-05', 32100.25),
    (4, '高橋', '美咲', 'takahashi.misaki@example.com', 31, '福岡', '2022-04-10', 15600.00),
    (5, '伊藤', '健太', 'ito.kenta@example.com', 39, '札幌', '2022-05-22', 28900.50),
    (6, '渡辺', '優子', 'watanabe.yuko@example.com', 26, '京都', '2022-06-18', 12450.75),
    (7, '山本', '大輔', 'yamamoto.daisuke@example.com', 45, '神戸', '2022-07-30', 42300.00),
    (8, '中村', '麗子', 'nakamura.reiko@example.com', 33, '仙台', '2022-08-14', 19850.25),
    (9, '小林', '隆', 'kobayashi.takashi@example.com', 50, '広島', '2022-09-05', 36700.50),
    (10, '加藤', '真理', 'kato.mari@example.com', 29, '横浜', '2022-10-12', 22100.75);

-- データの確認
SELECT * FROM customer_sample;
```

![image-20250427114927425.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1002774/c7defb89-e550-4973-9864-bd861d40f950.png)





# Redshift Serverless

Redshift Serverless を適当に作成したあと、Query editor を開いて、テーブルを作成します。

![image-20250427115144876.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1002774/d70e46a0-fa56-4178-9ca2-96d825c920da.png)



以下のテーブルを作成します。

```
CREATE TABLE IF NOT EXISTS dev.public.customer_sample (
    customer_id INTEGER,
    first_name VARCHAR(50),
    last_name VARCHAR(50),
    email VARCHAR(100),
    age INTEGER,
    city VARCHAR(50),
    registration_date DATE,
    total_purchase DECIMAL(10,2)
);
```

![image-20250427115254354.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1002774/7fbd0a90-ca76-45d7-9456-942620038bce.png)





# Glue : Snowflake Connection

AWS Glue で Snowflake Connection を作成していきます。Snowflake に接続するための文字列を AWS Secrets Manager に保存します。

Store a new secret を押します。

![image-20250427115602360.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1002774/5dde0990-f53a-47d1-9c83-956209520ee7.png)



以下に、Snowflake 接続用の Username を Password を入れます。

![image-20250420114335562.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1002774/88aa6a83-281b-4d9e-a1c4-2a8d59b3892d.png)



AWS Glue の画面に変更して、Create Connection を押します。

![image-20250420113437426.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1002774/5a65216b-1dd3-47a7-8db9-2624c983bdde.png)



Snowflake を選択します。

![image-20250420113511901.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1002774/19f5df4c-b28b-4887-b93a-791a4553b083.png)



設定をするために、Snowflake の画面上で、`Account/Server URL` を確認します。

![image-20250420113742794.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1002774/85c23d91-06b0-4c45-bb32-32fd75f23142.png)



- Host : 確認した `Account/Server URL` を入力
- Port : 443 を入力
- AWS Secret : 上の手順で作成した Secrets Manager の Store 名を入力

![image-20250420114438356.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1002774/3a0488f6-fd6c-47b1-a157-ff70901fc43f.png)



Test connection を押します。

![image-20250420114813217.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1002774/5a71f5a7-ea3a-42ed-9123-8368606a59e2.png)



Success と表示されれば OK です。

![image-20250420114822278.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1002774/0318dac0-fc39-4be7-b4f9-8cb6ecc672d6.png)



# Glue : Redshift Serverless 用 JDBC Conenction

Glue から Redshift Serverless に接続するために、Redshift Serverless の Workgroup configration から JDBC URL をメモします。

![image-20250427120356083.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1002774/74d805dd-bb8f-4584-94f3-97446e5f6dc3.png)



AWS Glue の画面に戻り、Create connection を選択します。

![image-20250427120501203.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1002774/f2e17538-a6f5-4842-a285-c3d55d12df03.png)



JDBC を選択します。

![image-20250427120543087.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1002774/3f1cd7bb-4d61-47e1-9eba-acc6473775ca.png)



- JDBC URL : 上記の手順で調べた文字列を指定
- AWS Secret : Redshift Severless の接続用 Username と Password を Store したものを指定。作成していない場合は、Username and password から指定可能
- Network options : Redshift Serverless を動かしている VPC や Subnet や Security Group を指定

![image-20250420143245156.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1002774/b6be793e-c641-42af-af5d-c6599ceaae63.png)



# Glue : Visual ETL

ここまで準備が出来たので、AWS Glue の Visual ETL を作成します。

Data source を Snowflake を指定します。Enter a custom query から、独自の Custom SQL を指定可能です。

> SELECT * FROM public.customer_sample WHERE registration_date >= '2022-05-01';

![image-20250420121610579.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1002774/62f4b0bb-067f-4dfc-bf44-07c73baa5317.png)



Target に Redshift Serverless を選択します。

- JDBC Connection を利用
- アクションは、以下の 4 つとなる。今回は MERGE を選択します。
  - APPEND (insert) to target table
    AWS Glueは既存のテーブルの列にデータを追加し、余分な列は破棄します。
    - APPEND は、単純に追記する。すでに Redshift 上に同じデータが存在していても、気にせず重複して追記をする。
  - MERGE data into target table
    AWS Glueは条件のセットに基づいて、テーブルにデータを更新または追加します。
  - TRUNCATE target table
    Appendと同様ですが、AWS Glueはまずテーブルの内容をクリアします。
  - DROP and recreate target table
    AWS Glueはテーブルを削除し、ソースデータからのスキーマでテーブルを再作成します。

![image-20250427121627603.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1002774/ff2592fd-f045-44e4-b282-97ed8d4e3374.png)



上記の設定をしたときに、Visual ETL の裏側で自動作成される PySpark のコードを確認できます。

![image-20250427121719603.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1002774/92188552-0166-444d-bca7-7c4bfeb4fc50.png)



以下のコードが見えます。

```python
import sys
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job
import gs_null_rows
from awsglue import DynamicFrame

args = getResolvedOptions(sys.argv, ['JOB_NAME'])
sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args['JOB_NAME'], args)

# Script generated for node Snowflake
Snowflake_node1745116384305 = glueContext.create_dynamic_frame.from_options(connection_type="snowflake", connection_options={"autopushdown": "on", "dbtable": "CUSTOMER_SAMPLE", "connectionName": "Snowflake connection", "sfDatabase": "TEST_DATABASE", "sfSchema": "PUBLIC"}, transformation_ctx="Snowflake_node1745116384305")

# Script generated for node Remove Null Rows
RemoveNullRows_node1745127959082 = Snowflake_node1745116384305.gs_null_rows()

# Script generated for node Amazon Redshift
AmazonRedshift_node1745127828495 = glueContext.write_dynamic_frame.from_options(frame=RemoveNullRows_node1745127959082, connection_type="redshift", connection_options={"postactions": "BEGIN; MERGE INTO public.customer_sample USING public.customer_sample_temp_xk4c59 ON customer_sample.CUSTOMER_ID = customer_sample_temp_xk4c59.CUSTOMER_ID WHEN MATCHED THEN UPDATE SET CUSTOMER_ID = customer_sample_temp_xk4c59.CUSTOMER_ID, FIRST_NAME = customer_sample_temp_xk4c59.FIRST_NAME, LAST_NAME = customer_sample_temp_xk4c59.LAST_NAME, EMAIL = customer_sample_temp_xk4c59.EMAIL, AGE = customer_sample_temp_xk4c59.AGE, CITY = customer_sample_temp_xk4c59.CITY, REGISTRATION_DATE = customer_sample_temp_xk4c59.REGISTRATION_DATE, TOTAL_PURCHASE = customer_sample_temp_xk4c59.TOTAL_PURCHASE WHEN NOT MATCHED THEN INSERT VALUES (customer_sample_temp_xk4c59.CUSTOMER_ID, customer_sample_temp_xk4c59.FIRST_NAME, customer_sample_temp_xk4c59.LAST_NAME, customer_sample_temp_xk4c59.EMAIL, customer_sample_temp_xk4c59.AGE, customer_sample_temp_xk4c59.CITY, customer_sample_temp_xk4c59.REGISTRATION_DATE, customer_sample_temp_xk4c59.TOTAL_PURCHASE); DROP TABLE public.customer_sample_temp_xk4c59; END;", "redshiftTmpDir": "s3://aws-glue-assets-xxxxxxxxxxxx-ap-northeast-1/temporary/", "useConnectionProperties": "true", "dbtable": "public.customer_sample_temp_xk4c59", "connectionName": "Jdbc connection", "preactions": "CREATE TABLE IF NOT EXISTS public.customer_sample (CUSTOMER_ID DECIMAL, FIRST_NAME VARCHAR, LAST_NAME VARCHAR, EMAIL VARCHAR, AGE DECIMAL, CITY VARCHAR, REGISTRATION_DATE DATE, TOTAL_PURCHASE DECIMAL); DROP TABLE IF EXISTS public.customer_sample_temp_xk4c59; CREATE TABLE public.customer_sample_temp_xk4c59 (CUSTOMER_ID DECIMAL, FIRST_NAME VARCHAR, LAST_NAME VARCHAR, EMAIL VARCHAR, AGE DECIMAL, CITY VARCHAR, REGISTRATION_DATE DATE, TOTAL_PURCHASE DECIMAL);"}, transformation_ctx="AmazonRedshift_node1745127828495")

job.commit()
```



「Script generated for node Amazon Redshift」の箇所を見やすく整形します。

```python
connection_options = {
    # 前処理アクション
    "preactions": """
        CREATE TABLE IF NOT EXISTS public.customer_sample (
            CUSTOMER_ID DECIMAL, 
            FIRST_NAME VARCHAR, 
            LAST_NAME VARCHAR, 
            EMAIL VARCHAR, 
            AGE DECIMAL, 
            CITY VARCHAR, 
            REGISTRATION_DATE DATE, 
            TOTAL_PURCHASE DECIMAL
        ); 
        DROP TABLE IF EXISTS public.customer_sample_temp_xk4c59; 
        CREATE TABLE public.customer_sample_temp_xk4c59 (
            CUSTOMER_ID DECIMAL, 
            FIRST_NAME VARCHAR, 
            LAST_NAME VARCHAR, 
            EMAIL VARCHAR, 
            AGE DECIMAL, 
            CITY VARCHAR, 
            REGISTRATION_DATE DATE, 
            TOTAL_PURCHASE DECIMAL
        );
    """,
    
    # 後処理アクション
    "postactions": """
        BEGIN; 
        MERGE INTO public.customer_sample 
        USING public.customer_sample_temp_xk4c59 
        ON customer_sample.CUSTOMER_ID = customer_sample_temp_xk4c59.CUSTOMER_ID 
        WHEN MATCHED THEN UPDATE SET 
            CUSTOMER_ID = customer_sample_temp_xk4c59.CUSTOMER_ID, 
            FIRST_NAME = customer_sample_temp_xk4c59.FIRST_NAME, 
            LAST_NAME = customer_sample_temp_xk4c59.LAST_NAME, 
            EMAIL = customer_sample_temp_xk4c59.EMAIL, 
            AGE = customer_sample_temp_xk4c59.AGE, 
            CITY = customer_sample_temp_xk4c59.CITY, 
            REGISTRATION_DATE = customer_sample_temp_xk4c59.REGISTRATION_DATE, 
            TOTAL_PURCHASE = customer_sample_temp_xk4c59.TOTAL_PURCHASE 
        WHEN NOT MATCHED THEN INSERT VALUES (
            customer_sample_temp_xk4c59.CUSTOMER_ID, 
            customer_sample_temp_xk4c59.FIRST_NAME, 
            customer_sample_temp_xk4c59.LAST_NAME, 
            customer_sample_temp_xk4c59.EMAIL, 
            customer_sample_temp_xk4c59.AGE, 
            customer_sample_temp_xk4c59.CITY, 
            customer_sample_temp_xk4c59.REGISTRATION_DATE, 
            customer_sample_temp_xk4c59.TOTAL_PURCHASE
        ); 
        DROP TABLE public.customer_sample_temp_xk4c59; 
        END;
    """,
    
    # S3の一時ディレクトリ
    "redshiftTmpDir": "s3://aws-glue-assets-xxxxxxxxxxxx-ap-northeast-1/temporary/",
    
    # 接続プロパティを使用
    "useConnectionProperties": "true",
    
    # 対象テーブル（一時テーブル）
    "dbtable": "public.customer_sample_temp_xk4c59",
    
    # 使用する接続名
    "connectionName": "Jdbc connection"
}
```



Redshift にデータを入れる (MERGE) している箇所について、重要な内容をピックアップします。

- **Glue の内部処理**:
  - glueContext.write_dynamic_frame.from_options() メソッドを呼び出すと、Glue は内部的にデータを指定された dbtable（この場合は一時テーブル public.customer_sample_temp_xk4c59）に挿入します。
- **実行順序**:
  - まず preactions で定義された SQL が実行され、テーブル構造が準備されます
  - 次に Glue が自動的に DynamicFrame のデータを一時テーブルに挿入します（これは内部で COPY コマンドや JDBC バッチ挿入などを使用）
  - 最後に postactions で定義された SQL が実行され、MERGE 操作などが行われます
- **データ転送の仕組み**:
  - Glue は redshiftTmpDir で指定された S3 バケットを一時ストレージとして使用
  - DynamicFrame のデータを S3 に一時的に書き出し
  - Redshift の COPY コマンドを使用して S3 からデータをロード（内部的に実行）



また、Glue Visual ETL では、独自の MERGE コマンドの指定も可能です。

![image-20250420150950137.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1002774/c4929c96-2d4e-4754-8b01-39bda6e143a1.png)



# 動作確認

Run を押します。

![image-20250427122236630.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1002774/1a473e36-6e87-4069-a1f2-be324a9bec46.png)



実行履歴を Runs タブから確認でき、1 分 28 秒で Succeeded となっていることがわかります。

![image-20250427122534817.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1002774/99888b89-854b-488f-adad-a47cc739d977.png)



実際に Redshift Serverless で SELECT してみると、Snowflake 上のデータが新規作成されていることがわかります。

![image-20250427122647559.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1002774/0897d93e-a3e1-4fcd-bcd2-c37033bc1ce5.png)



Snowflake 上のデータを更新してみましょう。健太から、誠に更新しました。

```
UPDATE customer_sample
SET first_name = '伊藤', last_name = '誠'
WHERE customer_id = 5;
```

![image-20250427122741835.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1002774/7fcb54cf-fa31-4baa-a4b4-fe4320a71f3e.png)



再び Glue を実行します。MERGE コマンドを利用しているので、変更されたところだけ更新されるはずです。

![image-20250427122815036.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1002774/195d49f7-7575-41ce-9b2c-34e389fc9f51.png)



想定通り、Redshift 上で更新されています。

![image-20250427124021019.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1002774/858dab98-b3cf-4444-a54a-0119b75dafcf.png)