GlueのApache Icebergベースマテリアライズドビューの特徴まとめと、S3 Tables Catalog上での構成手順

SELECT * FROM watanabe_namespace."mvsampletable$snapshots"
ORDER BY committed_at DESC;

SELECT * FROM "s3tablescatalog/watanabetable01"."watanabe_namespace"."mvsampletable"
FOR TIMESTAMP AS OF TIMESTAMP '2025-12-27 07:50:00';

SELECT customer_name, COUNT(*) as order_count, SUM(amount) as total_amount 
FROM glue_catalog.sales.orders
GROUP BY customer_name

SELECT * FROM customer_summary

CREATE TABLE watanabe_test_db.sampletable_iceberg (
    id INT,
    name STRING,
    age INT
)
LOCATION 's3://cm-watanabe-gluedata/warehouse/watanabe_test_db/sampletable_iceberg/'
TBLPROPERTIES (
    'table_type' = 'ICEBERG'
);

INSERT INTO "AwsDataCatalog".watanabe_test_db.sampletable_iceberg (id, name, age)
VALUES 
    (1, 'User1', 35),....;

import sys
import logging
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from pyspark.sql import SparkSession
from awsglue.context import GlueContext
from awsglue.job import Job
from datetime import datetime

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger("create_mv_from_glue_catalog_job")

def main():
    # --- パラメータ設定 ---
    args = {
        'JOB_NAME': 'create_mv_from_glue_catalog_job',
        's3t_catalog_id': '<account-id>:s3tablescatalog/watanabetable01',  # S3 Tables Catalog ID
        's3t_warehouse_path': 's3://watanabe-namespace/',  # S3 Tables warehouse(S3バケットは存在しなくてもよい)
        'glue_account_id': '<account-id>',  # Glue Data Catalog用のアカウントID
        'glue_warehouse_path': 's3://cm-watanabe-gluedata/warehouse/',  # Glue Data Catalog warehouse
        'region': 'us-east-1',
    }

    # --- Spark/Glueコンテキストの初期化 ---
    logger.info("Sparkの設定を開始します。")

    # 公式ドキュメントのS3 Tablesバケット + クロスカタログ参照の設定例に基づく
    # 参考: https://docs.aws.amazon.com/ja_jp/glue/latest/dg/materialized-views.html#materialized-views-creating
    spark_conf = SparkSession.builder.appName(args['JOB_NAME']) \
        .config("spark.sql.extensions", "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions") \
        .config("spark.sql.catalog.glue_catalog", "org.apache.iceberg.spark.SparkCatalog") \
        .config("spark.sql.catalog.glue_catalog.type", "glue") \
        .config("spark.sql.catalog.glue_catalog.warehouse", args['glue_warehouse_path']) \
        .config("spark.sql.catalog.glue_catalog.glue.region", args['region']) \
        .config("spark.sql.catalog.glue_catalog.glue.id", args['glue_account_id']) \
        .config("spark.sql.catalog.glue_catalog.glue.account-id", args['glue_account_id']) \
        .config("spark.sql.catalog.glue_catalog.client.region", args['region']) \
        .config("spark.sql.catalog.glue_catalog.glue.lakeformation-enabled", "false") \
        .config("spark.sql.catalog.s3t_catalog", "org.apache.iceberg.spark.SparkCatalog") \
        .config("spark.sql.catalog.s3t_catalog.type", "glue") \
        .config("spark.sql.catalog.s3t_catalog.glue.id", args['s3t_catalog_id']) \
        .config("spark.sql.catalog.s3t_catalog.glue.account-id", args['glue_account_id']) \
        .config("spark.sql.catalog.s3t_catalog.glue.lakeformation-enabled", "false") \
        .config("spark.sql.catalog.s3t_catalog.warehouse", args['s3t_warehouse_path']) \
        .config("spark.sql.catalog.s3t_catalog.glue.region", args['region']) \
        .config("spark.sql.catalog.s3t_catalog.client.region", args['region']) \
        .config("spark.sql.defaultCatalog", "s3t_catalog") \
        .config("spark.sql.optimizer.answerQueriesWithMVs.enabled", "true") \
        .config("spark.sql.materializedViews.metadataCache.enabled", "true") \
        .config("spark.sql.optimizer.incrementalMVRefresh.enabled", "true") \
        .config("spark.sql.optimizer.incrementalMVRefresh.deltaThresholdCheckEnabled", "false") \
        .config("spark.sql.optimizer.answerQueriesWithMVs.decimalAggregateCheckEnabled", "false") \

    logger.info("Sparkの設定を完了しました。")

    sc = SparkContext.getOrCreate(conf=spark_conf.getOrCreate().sparkContext.getConf())
    glueContext = GlueContext(sc)
    spark = glueContext.spark_session
    job = Job(glueContext)
    job.init(args['JOB_NAME'], args)

    # --- Materialized View作成クエリ ---
    # ベーステーブル: glue_catalog.watanabe_test_db.sampletable_iceberg
    # MV作成先: s3t_catalog.watanabe_namespace.mvsampletable_iceberg
    logger.info("マテリアライズドビューの作成クエリを開始します。")
    create_mv_query = """
    CREATE MATERIALIZED VIEW s3t_catalog.watanabe_namespace.mvsampletable_iceberg
    SCHEDULE REFRESH EVERY 1 HOUR
    AS
    SELECT
        id,
        name,
        age
    FROM glue_catalog.watanabe_test_db.sampletable_iceberg
    WHERE age > 30
    """

    logger.info("マテリアライズドビューの作成を開始します。")
    logger.info(f"SQLクエリを実行します。\n{create_mv_query}")

    try:
        # --- SQLクエリの実行 ---
        spark.sql(create_mv_query)
    except Exception:
        logger.exception("マテリアライズドビューの作成に失敗しました。")
        raise
    else:
        logger.info("マテリアライズドビューの作成に成功しました。")
        job.commit()
        logger.info("Jobは完了しました。")

if __name__ == "__main__":
    main()

CREATE MATERIALIZED VIEW s3t_catalog.watanabe_namespace.mvsampletable_iceberg
SCHEDULE REFRESH EVERY 1 HOUR
AS
    SELECT
        id,
        name,
        age
    FROM glue_catalog.watanabe_test_db.sampletable_iceberg
    WHERE age > 30

SHOW VIEWS FROM s3t_catalog.watanabe_namespace

+------------------+----------------+-----------+
|namespace         |viewName        |isTemporary|
+------------------+----------------+-----------+
|watanabe_namespace|mvsampletable   |false      |
|watanabe_namespace|mvsampletable_02|false      |
|watanabe_namespace|mvsampletable_03|false      |
|watanabe_namespace|mvsampletable_04|false      |
+------------------+----------------+-----------+

DESCRIBE EXTENDED s3t_catalog.watanabe_namespace.mvsampletable_iceberg

SHOW TBLPROPERTIES s3t_catalog.watanabe_namespace.mvsampletable_iceberg