🐴
AWS Glue Python Shell×chDB×PyIcebergで実現する軽量ETLとDuckDB比較

2026/01/05に公開
2026/01/06
 はじめにAWS Community Builderのぺんぎん（@jitepengin）です。

前回の記事でAWS Lambda、chDB、PyIcebergの組み合わせによる軽量ETLを実装しました。

https://zenn.dev/penginpenguin/articles/9a1dcfe74bcf6b
今回はそのGlue Python Shell版として、chDBを使ってIcebergに書き込むETLを実装・解説したいと思います。

Glue Python ShellはLambdaとは異なる特徴があります。こちらの特徴を活かしつつ、実装できればと思います。

また、前回の記事と同様に同条件でDuckDBと比較検証を行い、それぞれの特徴を探りたいと思います。

 AWS Glue Python ShellとはAWS Glue Python Shellは、Apache Sparkを使わずにPythonスクリプトを実行できるAWS Glueの実行環境です。

Lambdaの15分制限に比べると長時間のバッチ処理が可能になります。デフォルトではなんと48時間です！

また、サーバレスで起動できるため、インフラの管理は不要で使った分だけ課金されます。
GlueやEMRと比較するとコストも安く済むケースが多いことから、軽量ETLやデータ変換処理で活用されます。
そんな便利なGlue Python Shellですが、Lambdaと比べるとトリガとなるイベントが少ないというデメリットがあります。

よくイベント駆動の処理で使われるS3のファイル配置トリガについても直接のトリガやEventBridge経由でのトリガーも対応していません。

こういった場合のトリガーには一般的にはLambdaまたはGlue Workflowを経由して起動されます。

これ以外にも様々な制約がありますが、うまく使いこなせば処理の幅が大きく広がる可能性を秘めたサービスとなります。

 今回のアーキテクチャ
今回のアーキテクチャはとてもシンプルです。（DuckDBの記事と同じ構成にしています。）

S3にファイルがアップロードされたことをトリガーに、Lambdaを経由してGlue Python Shellを実行し、変換後のデータをApache Iceberg形式でS3に保存します。

LambdaはあくまでもGlue Python Shellの起動だけ行い、ETLをGlue Python Shell側で行うことで処理時間の上限の15分を超える処理を行うことが狙いとなります。
ポイントはLambdaで使用する各種ライブラリとなります。

chDB：データベースエンジンとして、メモリ内での高速なSQLクエリ処理を実現。

PyArrow：Arrowフォーマットを使用し、高速なデータの変換・転送をサポート。

PyIceberg：AWS Glue Catalogを活用して、Iceberg形式のテーブルにアクセス・操作。
ちなみに、Glue Python Shellの起動方法にはほかに、EventBridge経由でGlue Workflowを起動し、そこから、Glue Python Shellを起動する方法があります。

この起動方法の違いやメリットについては次の記事でまとめたいと思います。

 chDBとはhttps://clickhouse.com/chdb

chDBはClickHouseのクエリエンジンをライブラリとして組み込み利用可能にしたエンジンです。

主な特徴は以下の通りです。
ClickHouse の実行エンジンを内包
C++ネイティブ実装による高い実行性能
Pythonから直接SQLを実行可能
サーバーレス・シングルプロセスで利用可能
DuckDBと同様に「組み込み型分析エンジン」という位置付けです。

 Additional Python modules pathの設定Additional Python modules pathには以下を設定してください。
pyiceberg[glue]==0.9.1,pydantic==2.5.0,pydantic-core==2.14.1,annotated-types==0.6.0,chdb==0.13.0,pyarrow==14.0.1,typing-extensions==4.8.0

 サンプルコード
 Lambdaのトリガー処理S3のイベントを受けてGlue Python Shellを起動するだけのシンプルな処理です。
sample.py
import boto3

def lambda_handler(event, context):
    glue = boto3.client("glue")
    
    s3_bucket = event['Records'][0]['s3']['bucket']['name']
    s3_object_key = event['Records'][0]['s3']['object']['key']
            
    s3_input_path = f"s3://{s3_bucket}/{s3_object_key}"
    
    response = glue.start_job_run(
        JobName="chdb shell",
        Arguments={
            "--s3_input": s3_input_path
        }
    )
    print(f"Glue Job started: {response['JobRunId']}")
    return response

 AWS Glue Python ShellのETL処理Lambdaから起動され、パラメータをもとにchDBでファイルを取得、PyIcebergでS3のIcebergテーブルに書き込む処理です。

chDBではVendorID = 1でデータをフィルタリングしています。このようにSQLのような構文でクレンジングやフィルタリングを行うことが可能です。
今回もDuckDB版と同様にカタログアクセスにGlueCatalog形式を使っています。

AWS Glue Iceberg RESTエンドポイント方式（REST Catalog）でも同様の処理が可能となりますので、要件に応じて選択いただければと思います。
https://zenn.dev/penginpenguin/articles/e44880aaa2d5e3
sample.py
import boto3
import sys
import os
from awsglue.utils import getResolvedOptions


def get_job_parameters():
    try:
        required_args = ['s3_input']
        args = getResolvedOptions(sys.argv, required_args)

        s3_file_path = args['s3_input']
        print(f"s3_input: {s3_file_path}")

        return s3_file_path

    except Exception as e:
        print(f"parameters error: {e}")
        raise


def _to_pyarrow_table(result):
    """
    Compatibility helper to extract a pyarrow.Table from a chDB query_result.
    """
    import chdb

    if hasattr(chdb, "to_arrowTable"):
        return chdb.to_arrowTable(result)

    if hasattr(result, "to_pyarrow"):
        return result.to_pyarrow()
    if hasattr(result, "to_arrow"):
        return result.to_arrow()

    raise RuntimeError(
        "Cannot convert chdb query_result to pyarrow.Table. "
        f"Available attributes: {sorted(dir(result))[:200]}"
    )


def normalize_arrow_for_iceberg(table):
    """
    Normalize Arrow schema for Iceberg compatibility.
    - timestamptz -> timestamp
    """
    import pyarrow as pa

    new_fields = []
    new_columns = []

    for field, column in zip(table.schema, table.columns):
        # timestamp with timezone -> timestamp
        if pa.types.is_timestamp(field.type) and field.type.tz is not None:
            new_type = pa.timestamp(field.type.unit)
            new_fields.append(pa.field(field.name, new_type, field.nullable))
            new_columns.append(column.cast(new_type))
        else:
            new_fields.append(field)
            new_columns.append(column)

    new_schema = pa.schema(new_fields)
    return pa.Table.from_arrays(new_columns, schema=new_schema)


def read_parquet_with_chdb(s3_input):
    """
    Read Parquet file from S3 using chDB.
    """
    import chdb

    if s3_input.startswith("s3://"):
        bucket, key = s3_input.replace("s3://", "").split("/", 1)
        s3_url = f"https://{bucket}.s3.ap-northeast-1.amazonaws.com/{key}"
    else:
        s3_url = s3_input

    print(f"Reading data from S3: {s3_url}")

    query = f"""
        SELECT *
        FROM s3('{s3_url}', 'Parquet')
        WHERE VendorID = 1
    """

    result = chdb.query(query, "Arrow")
    arrow_table = _to_pyarrow_table(result)

    print("Original schema:")
    print(arrow_table.schema)

    # Normalize schema for Iceberg compatibility
    arrow_table = normalize_arrow_for_iceberg(arrow_table)

    print("Normalized schema:")
    print(arrow_table.schema)
    print(f"Rows: {arrow_table.num_rows:,}")

    return arrow_table


def write_iceberg_table(arrow_table):
    """
    Write Arrow table to Iceberg table using PyIceberg.
    """
    try:
        print("Writing started...")

        from pyiceberg.catalog import load_catalog

        catalog_config = {
            "type": "glue",
            "warehouse": "s3://your-bucket/your-warehouse/",  # Adjust to your environment.
            "region": "ap-northeast-1",
        }

        catalog = load_catalog("glue_catalog", **catalog_config)
        table_identifier = "icebergdb.yellow_tripdata"

        table = catalog.load_table(table_identifier)

        print(f"Target data to write: {arrow_table.num_rows:,} rows")
        table.append(arrow_table)

        return True

    except Exception as e:
        print(f"Writing error: {e}")
        import traceback
        traceback.print_exc()
        return False


def main():
    try:
        import chdb
        import pyiceberg

        # Read input parameter
        s3_input = get_job_parameters()

        # Read data with chDB
        arrow_tbl = read_parquet_with_chdb(s3_input)
        print(f"Data read success: {arrow_tbl.num_rows:,} rows")

        # Write to Iceberg table
        if write_iceberg_table(arrow_tbl):
            print("\nWriting fully successful!")
        else:
            print("Writing failed")

    except Exception as e:
        print(f"Main error: {e}")
        import traceback
        traceback.print_exc()


if __name__ == "__main__":
    main()

!今回は操作方法の違いにフォーカスしているので、Icebergテーブルのバージョン更新や競合によるエラー対応は割愛しています。

 実行結果
このようにIceberg形式のテーブルにデータを登録することができました！

 DuckDB版との比較以前紹介したDuckDB版と比較してみたいと思います。

DuckDB版の詳細は下記の記事を参考にどうぞ。

https://zenn.dev/penginpenguin/articles/5072c1270f33f2

 DuckDB版サンプルコードsample.py
import boto3
import sys
import os
from awsglue.utils import getResolvedOptions


def get_job_parameters():
    # get job parameters
    try:
        required_args = ['s3_input']
        
        args = getResolvedOptions(sys.argv, required_args)
        
        s3_file_path = args['s3_input']
        print(f"object: {s3_file_path}")

        return s3_file_path
        
    except Exception as e:
        print(f"parameters error: {e}")
        raise

def setup_duckdb_environment():
    # Properly set up the DuckDB environment
    try:
        home_dir = '/tmp'
        duckdb_dir = '/tmp/.duckdb'
        extensions_dir = '/tmp/.duckdb/extensions'
        
        os.environ['HOME'] = home_dir
        os.environ['DUCKDB_HOME'] = duckdb_dir
        os.environ['DUCKDB_CONFIG_PATH'] = duckdb_dir
        os.environ['DUCKDB_EXTENSION_DIRECTORY'] = extensions_dir
        
        os.makedirs(duckdb_dir, exist_ok=True)
        os.makedirs(extensions_dir, exist_ok=True)
        os.chmod(duckdb_dir, 0o755)
        os.chmod(extensions_dir, 0o755)
        
        print(f"DuckDB environment setup completed: {duckdb_dir}")
        return True
        
    except Exception as e:
        print(f"DuckDB environment setup error: {e}")
        return False

def read_parquet_with_duckdb(s3_input):
    # Read Parquet file from S3 using DuckDB
    import duckdb
    
    con = duckdb.connect(':memory:')
    
    try:
        con.execute("SET extension_directory='/tmp/.duckdb/extensions';")
        con.execute("INSTALL httpfs;")
        con.execute("LOAD httpfs;")
        
        session = boto3.Session()
        credentials = session.get_credentials()
        
        con.execute(f"SET s3_region='ap-northeast-1';")
        con.execute(f"SET s3_access_key_id='{credentials.access_key}';")
        con.execute(f"SET s3_secret_access_key='{credentials.secret_key}';")
        
        if credentials.token:
            con.execute(f"SET s3_session_token='{credentials.token}';")
        
        print(f"Reading data from S3: {s3_input}")
        sql = f"SELECT * FROM read_parquet('{s3_input}') WHERE VendorID = 1"
        res = con.execute(sql)
        
        return res.arrow()
        
    except Exception as e:
        print(f"DuckDB error: {e}")
        raise
    finally:
        con.close()

def write_iceberg_table(arrow_table):
    # Write to an Iceberg table
    try:
        print("Writing started...")
        
        from pyiceberg.catalog import load_catalog
        from pyiceberg.schema import Schema
        from pyiceberg.types import NestedField, StringType, IntegerType, DoubleType, TimestampType
        
        catalog_config = {
            "type": "glue",
            "warehouse": "s3://your-bucket/your-warehouse/", # Adjust to your environment.
            "region": "ap-northeast-1"
        }
        
        catalog = load_catalog("glue_catalog", **catalog_config)

        table_identifier = "icebergdb.yellow_tripdata"
        
        table = catalog.load_table(table_identifier)
        print(f"Target data to write: {arrow_table.num_rows:,} rows")
        
        # Write Arrow table to Iceberg table
        table.append(arrow_table)
        
     
        return True
        
    except Exception as e:
        print(f"Writing error: {e}")
        import traceback
        traceback.print_exc()
        return False

def main():  
    # Set up DuckDB environment
    if not setup_duckdb_environment():
        print("Failed to set up DuckDB environment")
        return
    
    try:
        import pyiceberg
        
        # Read data
        s3_input = get_job_parameters()

        arrow_tbl = read_parquet_with_duckdb(s3_input)
        print(f"Data read success: {arrow_tbl.shape}")
        
        # Write to Iceberg table
        if write_iceberg_table(arrow_tbl):
            print("\nWriting fully successful!")
  
        else:
            print("Writing failed")
        
    except Exception as e:
        print(f"Main error: {e}")
        import traceback
        traceback.print_exc()

if __name__ == "__main__":
    main()
!今回は操作方法の違いにフォーカスしているので、Icebergテーブルのバージョン更新や競合によるエラー対応は割愛しています。

 比較検証今回はNYCタクシーのデータを使用して同一ファイルをもとに、メモリを増減させた場合の比較を実施してみます。
https://www.nyc.gov/site/tlc/about/tlc-trip-record-data.page
ファイルは2024年のYellow Taxi Trip Recordsの1月分のファイル（2964624レコード、48MB）と

1年間のデータをまとめたファイル（41169720レコード、807MB）をテストデータとして扱います。
パラメータとしてDPUを1/16,1とした場合で検証します。

 検証結果
 1月分ファイル＝48MB

DPU
chDB 処理時間 (s)
chDB DPU hours
DuckDB 処理時間 (s)
DuckDB DPU hours


1/16
46
0.01未満
40
0.01未満

1
39
0.02
34
0.01


 1年分ファイル＝807MB

DPU
chDB 処理時間 (s)
chDB DPU hours
DuckDB 処理時間 (s)
DuckDB DPU hours


1/16
OutOfMemory
-
OutOfMemory
-

1
51
0.02
212
0.06

Glue Python Shellのメモリ割当がParquet→Arrow→Iceberg変換の一時バッファを処理できないためOutOfMemoryとなっている。

 考察小容量（約48MB）データでは chDB / DuckDB ともに十分実用的。差異は誤差レベル。
1DPUと十分なメモリを与えた場合、chDBは処理時間・コスト効率ともに優位。大きめのファイルを一気に処理するような用途ではchDBの特性が活きる。
Glue Python Shellによる軽量ETLではファイルサイズに応じたDPU設定が最重要！

 まとめ今回はAWS Glue Python ShellとchDBを活用してIceberg形式のテーブルに対してETL処理を実装しました。

特にDPU設定を1DPUにした場合かつ、大きいファイルを処理した際には処理時間・コスト効率ともに優位に働くことが分かりました。

前回の記事と同様に複雑なクエリではもしかすると傾向が異なる可能性がありますので、使用する際にはパフォーマンスの確認が必要となります。
Glue Python Shellの長時間実行可能な特性を活用することで、Lambdaではタイムアウトとなってしまうような処理も比較的コスト効率を高く実行することが可能となります。
また、今回紹介したようなchDBとDuckDBを活用することで、さらに効率的な処理を実現でき、PyIcebergを活用することでIceberg連携も可能となります。

OTFの主流とも言える存在となったIcebergを活用することでレイクハウスとして大きく活用範囲が広がります。
今回の記事が、Icebergテーブルを扱う際の軽量なデータ処理やリアルタイムETLを検討している方の参考になれば幸いです。
はじめに

AWS Glue Python Shellとは

今回のアーキテクチャ

chDBとは

Additional Python modules pathの設定

サンプルコード

Lambdaのトリガー処理

AWS Glue Python ShellのETL処理

実行結果

DuckDB版との比較

DuckDB版サンプルコード

比較検証

検証結果

1月分ファイル＝48MB

1年分ファイル＝807MB

考察

まとめ

Discussion
