如何在AWS Glue脚本中按主键查询DynamoDB并生成动态帧？

核心问题解答 #

AWS Glue原生的DynamoDB连接（通过create_dynamic_frame.from_options）不支持直接基于主键查询，它默认通过全表扫描或DDB导出功能获取全量数据，无法指定主键条件做过滤。

替代实现方案 #

如果要基于主键查询并转换成动态帧，可以按以下步骤操作：

• 用boto3的get_item或batch_get_item（注意batch_get_item单次最多返回100条，超过需分页处理）查询目标主键数据
• 将DDB格式的查询结果转换为PySpark DataFrame
• 再将DataFrame转为Glue动态帧

示例代码 #

import boto3
from pyspark.sql import SparkSession
from awsglue.dynamicframe import DynamicFrame

# 初始化DDB客户端
dynamodb = boto3.client('dynamodb')

# 定义要查询的主键列表（示例）
primary_keys = [
    {"id": {"S": "item1"}},
    {"id": {"S": "item2"}}
    # 单次最多100条，超过需分批处理
]

# 批量查询数据
response = dynamodb.batch_get_item(
    RequestItems={
        "your-ddb-table-name": {
            "Keys": primary_keys
        }
    }
)

# 转换DDB数据格式为Spark可识别的字典
def ddb_item_to_dict(item):
    def convert_value(val):
        if 'S' in val:
            return val['S']
        elif 'N' in val:
            return int(val['N'])
        elif 'BOOL' in val:
            return val['BOOL']
        # 根据实际数据类型扩展转换逻辑
        else:
            return str(val)
    return {k: convert_value(v) for k, v in item.items()}

items = [ddb_item_to_dict(item) for item in response['Responses']['your-ddb-table-name']]

# 获取Spark会话
spark = SparkSession.builder.getOrCreate()

# 转为DataFrame再转动态帧
df = spark.createDataFrame(items)
dynamic_frame = DynamicFrame.fromDF(df, glueContext, "pk_filtered_dynamic_frame")

注意事项 #

• 若主键数量超过100，需循环调用batch_get_item，处理返回的UnprocessedKeys直到无未处理键
• 转换DDB数据类型时，需根据实际存储的类型（如S、N、L、M等）完善转换函数，避免数据格式丢失
• 该方式适合小批量主键查询，若为大范围主键过滤，建议先将DDB数据导出到S3再用Glue处理，性能更优

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