基于Amazon Glue优化Iceberg表的合并功能在亚马逊云科技中国区域的实现

aws glue create-job \
    --name "temp-data-generator" \
    --role "<Your-Glue-Job-Role-Arn>" \
    --command '{ "Name": "glueetl", "ScriptLocation": "s3://<Your-Warehouse-Bucket-Name>/scripts/create_small_files.py", "PythonVersion": "3" }' \
    --glue-version "4.0" \
    --worker-type "G.1X" \
    --number-of-workers 2 \
    --default-arguments '{
        "--datalake-formats": "iceberg",
        "--conf": "spark.sql.catalog.glue_catalog=org.apache.iceberg.spark.SparkCatalog --conf spark.sql.catalog.glue_catalog.warehouse=s3://<Your-Warehouse-Bucket-Name>/warehouse/ --conf spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions --conf spark.sql.catalog.glue_catalog.catalog-impl=org.apache.iceberg.aws.glue.GlueCatalog --conf spark.sql.catalog.glue_catalog.io-impl=org.apache.iceberg.aws.s3.S3FileIO",
        "--database_name": "sales_db",
        "--table_name": "transactions",
        "--s3_warehouse_path": "s3://<Your-Warehouse-Bucket-Name>/warehouse/"
    }'

aws glue start-job-run --job-name "temp-data-generator"

aws glue create-job \
    --name "cleanup-snapshots-job" \
    --role "<Your-Glue-Job-Role-Arn>" \
    --command '{ "Name": "glueetl", "ScriptLocation": "s3://<Your-Warehouse-Bucket-Name>/scripts/cleanup_snapshots.py", "PythonVersion": "3" }' \
    --glue-version "4.0" \
    --worker-type "G.1X" \
    --number-of-workers 2 \
    --default-arguments '{
        "--datalake-formats": "iceberg",
        "--conf": "spark.sql.catalog.glue_catalog=org.apache.iceberg.spark.SparkCatalog --conf spark.sql.catalog.glue_catalog.warehouse=s3://<Your-Warehouse-Bucket-Name>/warehouse/ --conf spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions --conf spark.sql.catalog.glue_catalog.catalog-impl=org.apache.iceberg.aws.glue.GlueCatalog --conf spark.sql.catalog.glue_catalog.io-impl=org.apache.iceberg.aws.s3.S3FileIO",
        "--database_name": "sales_db",
        "--table_name": "transactions"
    }'

iceberg-compaction-template-minimal.yaml

AWSTemplateFormatVersion: '2010-09-09'
Description: 'A minimal, working solution for scheduled Iceberg table compaction using Amazon Glue and Lambda.'

Parameters:
  ProjectName:
    Type: String
    Default: 'iceberg-compaction-minimal'
    Description: 'A prefix for all created resources to ensure uniqueness.'
  DatabaseName:
    Type: String
    Default: 'iceberg_db'
    Description: 'The name of the Amazon Glue database for your Iceberg tables.'
  TableName:
    Type: String
    Default: 'my_iceberg_table'
    Description: 'The name of the Iceberg table you want to compact.'
  CompactionSchedule:
    Type: String
    Default: 'cron(0 2 ? * SUN *)' # 默认北京时间周日上午10点
    Description: 'The cron schedule for running the compaction job (in UTC).'

Resources:
  # 1. 创建用于Iceberg数据仓库和脚本的S3桶
  WarehouseBucket:
    Type: AWS::S3::Bucket
    Properties:
      BucketName: !Sub '${ProjectName}-warehouse-${AWS::AccountId}-${AWS::Region}'
      PublicAccessBlockConfiguration:
        BlockPublicAcls: true
        BlockPublicPolicy: true
        IgnorePublicAcls: true
        RestrictPublicBuckets: true

  # 2. 创建用于运行Glue Job的IAM Role
  GlueJobRole:
    Type: AWS::IAM::Role
    Properties:
      AssumeRolePolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Effect: Allow
            Principal: { Service: 'glue.amazonaws.com' }
            Action: 'sts:AssumeRole'
      ManagedPolicyArns:
        - !Sub 'arn:${AWS::Partition}:iam::aws:policy/service-role/AWSGlueServiceRole'
      Policies:
        - PolicyName: S3AccessPolicy
          PolicyDocument:
            Version: '2012-10-17'
            Statement:
              - Effect: Allow
                Action: 's3:*'
                Resource:
                  - !GetAtt WarehouseBucket.Arn
                  - !Sub '${WarehouseBucket.Arn}/*'

  # 3. 创建用于运行Lambda函数的IAM Role
  LambdaExecutionRole:
    Type: AWS::IAM::Role
    Properties:
      AssumeRolePolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Effect: Allow
            Principal: { Service: 'lambda.amazonaws.com' }
            Action: 'sts:AssumeRole'
      ManagedPolicyArns:
        - !Sub 'arn:${AWS::Partition}:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole'
      Policies:
        - PolicyName: GlueJobStartPolicy
          PolicyDocument:
            Version: '2012-10-17'
            Statement:
              - Effect: Allow
                Action: 'glue:StartJobRun'
                Resource: !Sub 'arn:${AWS::Partition}:glue:${AWS::Region}:${AWS::AccountId}:job/${CompactionJob}'

  # 4. 创建Glue Database存放Iceberg table表结构
  GlueDatabase:
    Type: AWS::Glue::Database
    Properties:
      CatalogId: !Ref AWS::AccountId
      DatabaseInput:
        Name: !Ref DatabaseName

  # 5. 创建Glue Job用来实现Compaction功能
  CompactionJob:
    Type: AWS::Glue::Job
    Properties:
      Name: !Sub '${ProjectName}-job'
      Role: !GetAtt GlueJobRole.Arn
      Command:
        Name: glueetl
        PythonVersion: '3'
        ScriptLocation: !Sub 's3://${WarehouseBucket}/scripts/iceberg_compaction.py'      
      DefaultArguments:
        '--datalake-formats': 'iceberg'
        '--conf': !Sub 'spark.sql.catalog.glue_catalog=org.apache.iceberg.spark.SparkCatalog --conf spark.sql.catalog.glue_catalog.warehouse=s3://${WarehouseBucket}/warehouse/ --conf spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions --conf spark.sql.catalog.glue_catalog.catalog-impl=org.apache.iceberg.aws.glue.GlueCatalog --conf spark.sql.catalog.glue_catalog.io-impl=org.apache.iceberg.aws.s3.S3FileIO'
        '--database_name': !Ref DatabaseName
        '--table_name': !Ref TableName
      GlueVersion: '4.0'
      WorkerType: 'G.1X'
      NumberOfWorkers: 2

  # 6. 创建Lambda触发器，来触发Glue Job
  CompactionTriggerFunction:
    Type: AWS::Lambda::Function
    Properties:
      FunctionName: !Sub '${ProjectName}-trigger'
      Runtime: python3.9
      Handler: index.lambda_handler
      Role: !GetAtt LambdaExecutionRole.Arn
      Timeout: 60
      Environment:
        Variables:
          GLUE_JOB_NAME: !Ref CompactionJob
      Code:
        ZipFile: |
          import boto3
          import os
          import logging

          logger = logging.getLogger()
          logger.setLevel(logging.INFO)
          glue_client = boto3.client('glue')

          def lambda_handler(event, context):
              job_name = os.environ['GLUE_JOB_NAME']
              logger.info(f"Starting Glue job: {job_name}")
              try:
                  response = glue_client.start_job_run(JobName=job_name)
                  logger.info(f"Successfully started job run. Run ID: {response['JobRunId']}")
                  return { 'statusCode': 200, 'body': f"Job {job_name} started." }
              except Exception as e:
                  logger.error(f"Failed to start job {job_name}: {e}")
                  raise e

  # 7. 创建EventBridge规则来定期执行Lambda函数
  ScheduleRule:
    Type: AWS::Events::Rule
    Properties:
      Name: !Sub '${ProjectName}-schedule'
      ScheduleExpression: !Ref CompactionSchedule
      State: ENABLED
      Targets:
        - Arn: !GetAtt CompactionTriggerFunction.Arn
          Id: 'CompactionTriggerTarget'

  # 8. 为Eventbridge创建相关权限来调用Lambda函数
  LambdaInvokePermission:
    Type: AWS::Lambda::Permission
    Properties:
      FunctionName: !GetAtt CompactionTriggerFunction.Arn
      Action: 'lambda:InvokeFunction'
      Principal: 'events.amazonaws.com'
      SourceArn: !GetAtt ScheduleRule.Arn

Outputs:
  WarehouseBucketName:
    Description: 'S3 bucket for the Iceberg warehouse and scripts.'
    Value: !Ref WarehouseBucket
  CompactionJobName:
    Description: 'Name of the created Amazon Glue job for compaction.'
    Value: !Ref CompactionJob
  LambdaFunctionName:
    Description: 'Name of the trigger Lambda function.'
    Value: !Ref CompactionTriggerFunction

#!/bin/bash

# 部署Iceberg表Compaction功能脚本

set -e # Exit immediately if a command exits with a non-zero status.

# 配置列表
STACK_NAME="my-iceberg-compaction-stack"
TEMPLATE_FILE="iceberg-compaction-template-minimal.yaml"
# You can specify your region here, or it will use your default configured region.
AWS_REGION="${AWS_REGION:-cn-north-1}" 

echo "Starting deployment of stack: ${STACK_NAME} in region: ${AWS_REGION}"

# 部署前检查验证模板文件
echo "Validating template: ${TEMPLATE_FILE}..."
aws cloudformation validate-template --template-body file://${TEMPLATE_FILE} > /dev/null
echo "Template validation successful."

# 部署Amazon Cloudformation堆栈，新建或更新堆栈
echo "Deploying CloudFormation stack..."
aws cloudformation deploy \
  --stack-name "${STACK_NAME}" \
  --template-file "${TEMPLATE_FILE}" \
  --capabilities CAPABILITY_NAMED_IAM \
  --region "${AWS_REGION}" \
  --parameter-overrides \
    ProjectName="my-iceberg-app" \
    DatabaseName="sales_db" \
    TableName="transactions"

echo "Deployment command sent. Waiting for stack operation to complete..."

# 输出堆栈outputs
echo "Deployment successful! Stack outputs:"
aws cloudformation describe-stacks \
  --stack-name "${STACK_NAME}" \
  --query "Stacks[0].Outputs" \
  --output table \
  --region "${AWS_REGION}"

echo "Next Steps: You need to upload your compaction Glue script to the created S3 bucket."

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 logging

# 设置日志
logging.basicConfig()
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)

# 主Spark任务

# 1. 初始化
args = getResolvedOptions(sys.argv, ['JOB_NAME', 'database_name', 'table_name'])
sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args['JOB_NAME'], args)

# 2. 从参数中获取数据库和表名
db_name = args['database_name']
table_name = args['table_name']
full_table_name = f"glue_catalog.{db_name}.{table_name}"

logger.info(f"Started Compaction operation on table '{full_table_name}'")

# 3. 执行核心合并逻辑
try:
    logger.info("Calling 'rewrite_data_files' Stored Procedure")
    result_df = spark.sql(f"""
        CALL glue_catalog.system.rewrite_data_files(
            table => '{db_name}.{table_name}', 
            strategy => 'binpack',
            # OPTIONS map can make the compaction more aggressive, ensuring it processes all small files.
            options => map(
                'min-input-files', '2',
                'target-file-size-bytes', '134217728'
            )
        )
    """)
    
    logger.info("'rewrite_data_files'Completed, and the results are as following:")
    result_df.show()

    result_data = result_df.first()
    if result_data:
        rewritten_files = result_data["rewritten_data_files_count"]
        added_files = result_data["added_data_files_count"]
        logger.info(f" Completed compaction on {rewritten_files} files. {added_files} new files generated.")
    else:
        logger.info(" No files need to be compacted. Current table is already optimized status.")

except Exception as e:
    logger.error(f"Compaction job failed: {e}")
    raise e

logger.info(f" Compaction operation on table '{full_table_name}' completed successfully")

job.commit()

import sys
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job
from pyspark.sql.types import StructType, StructField, StringType, IntegerType

# 初始化
args = getResolvedOptions(sys.argv, ['JOB_NAME', 'database_name', 'table_name'])
sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args['JOB_NAME'], args)

db_name = args['database_name']
table_name = args['table_name']
full_table_name_for_sql = f"glue_catalog.{db_name}.{table_name}"
full_table_name_for_save = f"glue_catalog.{db_name}.{table_name}" # In Glue 4/Iceberg 1.x, save() also needs the full name

print(f"Preparing to generate testing data in table '{full_table_name_for_sql}'.")

# 使用SQL DDL语句创建测试用的Iceberg表，其中TBLPROPERTIES参数为启用Compaction功能的前提，非常重要
try:
    print(f"Creating table {full_table_name_for_sql}")
    spark.sql(f"""
        CREATE TABLE IF NOT EXISTS {full_table_name_for_sql} (
            id int,
            name string
        )
        USING iceberg
        TBLPROPERTIES (
            'format-version'='2',
            'write.object-storage.enabled'='true'
        )
    """)
    print("Table created or exists")
except Exception as e:
    print(f"Failed when creating table: {e}")
    raise e

# 向创建后的表持续使用Append方式写入数据
schema = StructType([
    StructField("id", IntegerType(), True),
    StructField("name", StringType(), True)
])

for i in range(100):
    print(f"Writing the No. {i+1}/100 file …")
    data = [(i, f"user_{i}")]
    df = spark.createDataFrame(data, schema)
    
    df.write.format("iceberg").mode("append").save(full_table_name_for_save)

print("100 files created successfully!")
job.commit()

import sys
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job
import logging
from pyspark.sql.functions import current_timestamp

# 设置日志
logging.basicConfig()
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)

# 清理逻辑的主Spark Job

# 1. 初始化
args = getResolvedOptions(sys.argv, ['JOB_NAME', 'database_name', 'table_name'])
sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args['JOB_NAME'], args)

# 2. 获取参数
db_name = args['database_name']
table_name = args['table_name']
full_table_name = f"glue_catalog.{db_name}.{table_name}"

logger.info(f"Started snapshot cleanup operation on table '{full_table_name}' .")

# 3. 执行核心清理逻辑，调用Iceberg内置的expire_snapshots存储过程。这会删除所有比当前时间早的旧快照，然后自动删除那些只被这些过期快照引用的数据文件。
try:
    logger.info(" Calling Expire_Snapshots Stored Procedure ...")
    
    # 获取当前时间戳用于清理
    current_ts_df = spark.sql("SELECT current_timestamp()")
    current_ts = current_ts_df.collect()[0][0]
    
    result_df = spark.sql(f"""
        CALL glue_catalog.system.expire_snapshots(
            table => '{db_name}.{table_name}',
            older_than => TIMESTAMP '{current_ts}'
        )
    """)
    
    logger.info("Expire_Snapshots execution completed. Results are as following: ")
    result_df.show(truncate=False)

except Exception as e:
    logger.error(f" Snapshot cleanup job failed: {e}")
    raise e

logger.info(f" Snapshot cleanup operation on table '{full_table_name}' completed! ")

job.commit()