无需Glue Crawler，如何手动向Athena表插入新数据？

Athena支持INSERT语句，但对于你的场景（S3中持续新增JSON文件），更高效低成本的方案是管理表的元数据（分区或数据位置）——因为Athena是基于S3的无服务器查询服务，表本质是指向S3数据路径的元数据映射，数据始终存储在S3中，无需"插入"操作。

核心思路 #

Glue Crawler成本高的原因是它会定期全量扫描S3路径更新元数据，而我们可以通过Lambda触发，仅针对新增的文件/路径更新Athena表的元数据，避免无意义的全量扫描，大幅降低成本。

推荐方案：分区表（最优） #

如果你的S3数据按规则分区（比如按时间：year=YYYY/month=MM/day=DD/hour=HH/minute=MM，对应每20分钟的批次），可以通过Lambda执行ALTER TABLE ADD PARTITION语句，将新增的分区路径关联到Athena表。这种方式不仅成本低，还能提升后续查询的性能（减少扫描的数据量）。

备选方案：非分区表 #

如果没有分区，可通过MSCK REPAIR TABLE让Athena自动发现新增的文件，或用ALTER TABLE ADD LOCATION手动添加新的数据路径。但频繁执行MSCK会扫描整个表的S3路径，性能和成本不如分区表。

Python Lambda实现步骤 #

1. 前置准备 #

• 确保Athena表已创建（可复用之前Glue Crawler生成的表结构，或手动用CREATE TABLE语句创建）。
• 给Lambda的IAM角色添加以下权限：
  • S3只读访问（获取新增文件路径）
  • Athena查询执行权限
  • Glue Data Catalog访问权限（更新表元数据）
• 设置Athena查询结果的S3存储桶（用于存放查询执行日志）。

2. 分区表实现代码示例 #

以下Lambda函数监听S3的PutObject事件，提取新增文件所在的分区路径，自动添加到Athena表：

import boto3
import time

# 配置参数
ATHENA_DB = "your_database_name"
ATHENA_TABLE = "your_table_name"
ATHENA_OUTPUT_BUCKET = "s3://your-athena-results-bucket/"
TARGET_S3_BUCKET = "your-data-bucket"

athena = boto3.client("athena")

def run_athena_query(query):
    # 提交Athena查询
    resp = athena.start_query_execution(
        QueryString=query,
        QueryExecutionContext={"Database": ATHENA_DB},
        ResultConfiguration={"OutputLocation": ATHENA_OUTPUT_BUCKET}
    )
    query_id = resp["QueryExecutionId"]
    
    # 等待查询完成
    while True:
        status = athena.get_query_execution(QueryExecutionId=query_id)[
            "QueryExecution"
        ]["Status"]["State"]
        if status in ["SUCCEEDED", "FAILED", "CANCELLED"]:
            break
        time.sleep(2)
    
    if status == "FAILED":
        error = athena.get_query_execution(QueryExecutionId=query_id)[
            "QueryExecution"
        ]["Status"]["StateChangeReason"]
        raise Exception(f"Query failed: {error}")
    return query_id

def lambda_handler(event, context):
    for record in event["Records"]:
        # 提取新增文件的S3键
        s3_key = record["s3"]["object"]["key"]
        # 拆分路径，提取分区字段（如year=2024/month=05）
        path_segments = s3_key.split("/")
        partition_fields = [seg for seg in path_segments if "=" in seg]
        if not partition_fields:
            continue
        
        # 构造分区语句和存储路径
        partition_clause = ", ".join(
            [f"{k} = '{v}'" for k, v in [seg.split("=") for seg in partition_fields]]
        )
        partition_location = f"s3://{TARGET_S3_BUCKET}/{'/'.join(path_segments[:-1])}/"
        
        # 生成并执行ALTER语句（IF NOT EXISTS避免重复添加）
        alter_query = f"""
        ALTER TABLE {ATHENA_TABLE} 
        ADD IF NOT EXISTS PARTITION ({partition_clause}) 
        LOCATION '{partition_location}'
        """
        try:
            run_athena_query(alter_query)
            print(f"Added partition: {partition_clause}")
        except Exception as e:
            print(f"Error adding partition: {str(e)}")

3. 非分区表实现代码示例 #

如果没有分区，可在Lambda中执行MSCK REPAIR TABLE来刷新元数据：

# 复用上面的run_athena_query函数
def lambda_handler(event, context):
    msck_query = f"MSCK REPAIR TABLE {ATHENA_TABLE};"
    try:
        run_athena_query(msck_query)
        print("Refreshed table metadata successfully")
    except Exception as e:
        print(f"Error refreshing metadata: {str(e)}")

成本优势说明 #

• Lambda：每月前100万次调用免费，超出后每百万次仅0.2美元，你的场景下几乎无成本。
• Athena：执行ALTER或MSCK语句的扫描数据量极小（仅元数据），成本可以忽略不计，远低于Glue Crawler的小时计费模式。

注意事项 #

• Lambda超时时间：设置为5分钟足够覆盖Athena元数据操作的执行时间。
• 避免重复添加分区：使用ADD IF NOT EXISTS防止重复执行导致的错误。
• 批量处理：如果每20分钟有多个文件，可在Lambda中批量提取分区路径，合并为一条ALTER语句执行，减少Athena查询次数。

内容的提问来源于stack exchange，提问作者Annis99