[新機能] Amazon Redshift Data APIがCSV結果フォーマットをサポートしたので試してみました

石川覚
2024.11.03
AWS事業本部コンサルティング部の石川です。Amazon Redshift Data APIが新たにCSV結果フォーマットをサポートし、クエリ結果をCSV（カンマ区切り）形式で直接取得できるようになりました。この新機能がどのように便利なのか、実際に試してみます。また、CSVファイルを出力するサンプルコードを作成し、詳しく解説します。
https://aws.amazon.com/jp/about-aws/whats-new/2024/11/csv-result-format-amazon-redshift-data-api/
 Amazon Redshift Data APIがCSV結果フォーマットをサポートとはAmazon Redshift Data APIは、これまでJSONフォーマットでのみ結果を返していましたが、今回のアップデートでCSVフォーマットのサポートが追加されました。この新しいオプションにより、ユーザーは以下のメリットがあります。
データの可読性向上
スプレッドシートアプリケーションとの互換性
データ処理の効率化
 ユースケースこの新機能は、以下のようなユースケースで効果的です。
 データ分析CSVフォーマットはExcelなどのスプレッドシートソフトと親和性が高く、即座にデータ分析を開始できます。
 レポート作成定期的なレポート作成プロセスを自動化する際、CSVフォーマットは扱いやすいフォーマットです。
 データ連携他のシステムとのデータ連携において、CSVは広く受け入れられているフォーマットです。
 新しいパラメータとGetStatementResultV2 APIについて新しいCSVフォーマットを利用するには、API呼び出し時にResultFormatパラメータをCSVに設定するだけです。これにより、クエリ結果が直接CSVフォーマットで返されます。
例えば、AWS CLIを使用する場合は以下のようなコマンドになります。
aws redshift-data execute-statement \
    --cluster-identifier mycluster \
    --database dev \
    --sql "SELECT * FROM order LIMIT 10" \
    --result-format CSV
CSV結果フォーマットにより、ExecuteStatementおよびBatchExecuteStatement APIを呼び出す際に、--result-formatパラメータを通じてクエリ結果をJSONまたはCSVのフォーマットを指定できるようになりました。
CSV結果を取得するには、CSV結果をサポートする新しい GetStatementResultV2 APIを使用し、GetStatementResult APIは、引き続きJSONのみをサポートします。指定しない場合、デフォルトのフォーマットはJSONのままです。
 awsコマンド（awscli）でCSV結果フォーマットを取得するでは、サクッと awsコマンドを使って簡単に動作を確認します。なお、awsコマンド（awscli）は、執筆時点で**最新のaws-cli/2.19.1**です。
% aws --version
aws-cli/2.19.1 Python/3.12.7 Darwin/23.6.0 source/arm64
CSV形式で結果を取得するため、--result-format CSVを指定して、クエリを実行します。
% aws redshift-data execute-statement \
  --workgroup-name dev-wg \
  --database dev \
  --sql "SELECT now()" \
  --result-format CSV 

{
    "CreatedAt": "2024-11-03T21:34:38.603000+09:00",
    "Database": "dev",
    "DbUser": "IAMR:cm-ishikawa",
    "Id": "500cdf7c-9786-4188-b3f3-0cf91f5b0941",
    "WorkgroupName": "dev-wg"
}
上記のidを指定して、クエリの実行状況を取得します。
% aws redshift-data describe-statement \
  --id "500cdf7c-9786-4188-b3f3-0cf91f5b0941" 

{
    "CreatedAt": "2024-11-03T21:34:38.603000+09:00",
    "Database": "dev",
    "DbUser": "IAMR:cm-ishikawa",
    "Duration": 7387722,
    "HasResultSet": true,
    "Id": "500cdf7c-9786-4188-b3f3-0cf91f5b0941",
    "QueryString": "SELECT now()",
    "RedshiftPid": 1073815652,
    "RedshiftQueryId": 0,
    "ResultFormat": "csv",
    "ResultRows": 1,
    "ResultSize": 36,
    "Status": "FINISHED",
    "UpdatedAt": "2024-11-03T21:34:39.234000+09:00",
    "WorkgroupName": "dev-wg"
}
上記の実行ステイタス（Status）が、FINISHEDなので、idを指定してクエリの実行結果を取得します。CSV形式の実行結果を取得するためget-statement-result-v2を指定します。
RecordsのCSVRecordsのクエリの実行結果がエンコードされたCSV文字列として取得できたことを確認できます。
% aws redshift-data get-statement-result-v2 \
  --id "500cdf7c-9786-4188-b3f3-0cf91f5b0941" 

{
    "Records": [
        {
            "CSVRecords": "now\r\n2024-11-03 12:34:39.068393+00\r\n"
        }
    ],
    "ColumnMetadata": [
        {
            "isCaseSensitive": false,
            "isCurrency": false,
            "isSigned": false,
            "label": "now",
            "length": 0,
            "name": "now",
            "nullable": 1,
            "precision": 35,
            "scale": 6,
            "schemaName": "",
            "tableName": "",
            "typeName": "timestamptz"
        }
    ],
    "TotalNumRows": 1,
    "ResultFormat": "csv"
}  
 クエリの実行結果をCSVファイル出力するサンプルコードを実装ユースケースの例に挙げた「データ連携」想定して、クエリの実行結果をcsvファイルで取得するサンプルプログラム（Python/Boto3）を作成しました。
 Pythonモジュール（boto3）のアップデートプログラムを作成する前に、Pythonモジュール（boto3）を最新にアップデートしてください。なお、awsコマンド（awscli）は、執筆時点で最新の**boto3はboto3-1.35.54、botocoreはbotocore-1.35.54**です。
% pip install -U boto3
Requirement already satisfied: boto3 in /Users/ishikawa.satoru/.pyenv/versions/3.10.11/lib/python3.10/site-packages (1.35.38)
Collecting boto3
  Downloading boto3-1.35.54-py3-none-any.whl.metadata (6.7 kB)
Collecting botocore<1.36.0,>=1.35.54 (from boto3)
  Downloading botocore-1.35.54-py3-none-any.whl.metadata (5.7 kB)
Requirement already satisfied: jmespath<2.0.0,>=0.7.1 in /Users/ishikawa.satoru/.pyenv/versions/3.10.11/lib/python3.10/site-packages (from boto3) (1.0.1)
Requirement already satisfied: s3transfer<0.11.0,>=0.10.0 in /Users/ishikawa.satoru/.pyenv/versions/3.10.11/lib/python3.10/site-packages (from boto3) (0.10.0)
Requirement already satisfied: python-dateutil<3.0.0,>=2.1 in /Users/ishikawa.satoru/.pyenv/versions/3.10.11/lib/python3.10/site-packages (from botocore<1.36.0,>=1.35.54->boto3) (2.9.0.post0)
Requirement already satisfied: urllib3!=2.2.0,<3,>=1.25.4 in /Users/ishikawa.satoru/.pyenv/versions/3.10.11/lib/python3.10/site-packages (from botocore<1.36.0,>=1.35.54->boto3) (1.26.18)
Requirement already satisfied: six>=1.5 in /Users/ishikawa.satoru/.pyenv/versions/3.10.11/lib/python3.10/site-packages (from python-dateutil<3.0.0,>=2.1->botocore<1.36.0,>=1.35.54->boto3) (1.16.0)
Downloading boto3-1.35.54-py3-none-any.whl (139 kB)
Downloading botocore-1.35.54-py3-none-any.whl (12.7 MB)
   ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 12.7/12.7 MB 25.3 MB/s eta 0:00:00
Installing collected packages: botocore, boto3
  Attempting uninstall: botocore
    Found existing installation: botocore 1.35.38
    Uninstalling botocore-1.35.38:
      Successfully uninstalled botocore-1.35.38
  Attempting uninstall: boto3
    Found existing installation: boto3 1.35.38
    Uninstalling boto3-1.35.38:
      Successfully uninstalled boto3-1.35.38
Successfully installed boto3-1.35.54 botocore-1.35.54

[notice] A new release of pip is available: 24.2 -> 24.3.1
[notice] To update, run: pip install --upgrade pip
以下のプログラムは、Amazon Redshift Data APIを用いて、Amazon Redshift Serverlessにクエリを実行します。実行のステイタスがFINISHEDになると、実行結果を取得してCSVファイル（output.csv）出力します。
クエリの実行は、client.execute_statement()の引数にResultFormat='CSV'を新たに追加しました。
結果の取得は、従来のget_statement_result()ではなく、新しいget_statement_result_v2()を使って結果を取得します。
import boto3
import time
import csv
import io

def execute_redshift_serverless_query():

    # redshift-dataのclient作成
    PROFILE_NAME = 'ishikawa' # Optional
    session = boto3.Session()
    if PROFILE_NAME in boto3.Session().available_profiles:
        session = boto3.Session(profile_name=PROFILE_NAME)
    client = session.client('redshift-data')

    # Redshift Serverless接続情報
    workgroup_name = 'dev-wg'
    database = 'dev'

    # 実行するSQL
    sql = """
    SELECT 
        order_id, 
        order_date, 
        customer_id, 
        customer_name, 
        replace(product_name, ',', ' ') as product_name 
    FROM 
        dev.public.orders 
    LIMIT 5;
    """

    try:
        # SQLの実行
        response = client.execute_statement(
            WorkgroupName=workgroup_name,
            Database=database,
            Sql=sql,
            ResultFormat='CSV'
        )

        # 実行IDの取得
        query_id = response['Id']
        print(f"クエリID: {query_id}")

        # クエリの完了を待つ
        while True:
            # クエリのステータスを確認
            status_response = client.describe_statement(Id=query_id)
            status = status_response['Status']

            print(f"現在のステータス: {status}")

            if status == 'FINISHED':
                print("クエリが正常に完了しました。")
                break
            elif status in ['FAILED', 'ABORTED']:
                print(f"クエリが失敗しました。ステータス: {status}")
                if 'Error' in status_response:
                    print(f"エラー: {status_response['Error']}")
                return

            time.sleep(2)  # 2秒待機してから再確認

        # 結果の取得
        result = client.get_statement_result_v2(Id=query_id)

        # 結果の表示
        raw_csv = []
        for row in result['Records']:
            raw_csv.append(row['CSVRecords'])

        # StringIOを使用してデータを読み込む
        input_file = io.StringIO(''.join(raw_csv))

        # CSVリーダーを使用してデータを読み込む
        csv_reader = csv.reader(input_file)

        # 出力ファイル名
        output_file = 'output.csv'

        # CSVファイルに書き込む
        with open(output_file, 'w', newline='', encoding='utf-8') as file:
            csv_writer = csv.writer(file)
            for row in csv_reader:
                csv_writer.writerow(row)

        print(f"CSVファイルが '{output_file}' として出力されました。")

    except Exception as e:
        print(f"エラーが発生しました: {str(e)}")

if __name__ == "__main__":
    execute_redshift_serverless_query()
 実行結果出力されるCSVファイルは、ヘッダ（見出し行）がついたカンマ区切り形式のレコードファイルです。
% python run-query-get-statement-result-v2.py
クエリID: 881c3648-f9c3-4297-be16-2f2de1efcee9
現在のステータス: PICKED
現在のステータス: FINISHED
クエリが正常に完了しました。
CSVファイルが 'output.csv' として出力されました。

% cat output.csv
order_id,order_date,customer_id,customer_id,customer_name,product_name
CA-2016-152156,2016-11-08,CG-12520,CG-12520,Claire Gute,Bush Somerset Collection Bookcase
CA-2016-152156,2016-11-08,CG-12520,CG-12520,Claire Gute,Hon Deluxe Fabric Upholstered Stacking Chairs  Rounded Back
CA-2016-138688,2016-06-12,DV-13045,DV-13045,Darrin Van Huff,Self-Adhesive Address Labels for Typewriters by Universal
US-2015-108966,2015-10-11,SO-20335,SO-20335,Sean O'Donnell,Bretford CR4500 Series Slim Rectangular Table
US-2015-108966,2015-10-11,SO-20335,SO-20335,Sean O'Donnell,Eldon Fold 'N Roll Cart System
 最後にAmazon Redshift Data APIの新機能であるCSV結果フォーマットのサポートは、データ分析やシステム連携において大きな利点をもたらします。従来のJSONフォーマットに加えて、CSVフォーマットが選択できるようになったことで、ユーザーはより柔軟にデータを扱えるようになりました。特に、スプレッドシートアプリケーションとの互換性が向上し、データの可読性と処理効率が高まったことは注目に値します。
新しいGetStatementResultV2 APIを使用することで、CSV形式の結果を直接取得できるようになり、データ連携や定期的なレポート作成プロセスの自動化がより簡単になりました。サンプルコードで示したように、PythonとBoto3を使用して簡単にCSV形式の結果を取得し、ファイルに出力できることも、この新機能の実用性を認識していただけたはずです。
実際に使ってみて感じたことは、デリミタ（区切り文字）がカンマ以外に変更できないため、Redshiftが返す値にカンマを含んでいた場合にエンコードするなどの対策が必要になります。今回はカンマを空白に置き換えて対処しました。デリミタ（区切り文字）をパラメタ指定できるアップデートを期待します。
Amazon Redshift Data APIのCSV結果フォーマットサポートは、データウェアハウスの利用をより効率的かつ柔軟にする重要な進化であり、多くのユーザーにとって有益なアップデートとなるよう期待しています。