优化基于GlueContext Python从AWS DynamoDB获取记录的方案

核心问题分析 #

你的代码当前是先全表扫描DynamoDB表，再在Spark层面过滤department="Finance"的记录，这会读取表中所有数据，消耗大量读容量单位（RCU），导致不必要的成本开销。优化的关键是将过滤逻辑上移到DynamoDB服务端，只读取符合条件的数据。

最优实现思路：Query操作+全局二级索引（GSI） #

DynamoDB的Query操作是高效的定向读取方式，仅针对分区键（主表分区键或GSI分区键）匹配的数据进行读取，不会扫描全表。要实现按department过滤，需先为department列创建GSI（如果它不是主表的分区键）。

步骤1：创建全局二级索引（GSI） #

如果你的DynamoDB表未将department设为主表分区键，需创建GSI：

• 登录DynamoDB控制台，找到目标表
• 进入「索引」标签，点击「创建索引」
• 选择「全局二级索引」，设置：
  • 分区键：department（字符串类型）
  • 索引名称：比如department-index（自定义名称）
  • 按需配置读写容量（建议设为按需模式，降低成本）

步骤2：修改Glue代码实现定向查询 #

修改代码，使用Query操作通过GSI直接获取department="Finance"的记录，避免全表扫描：

def fetch_data(table_name):
    # 通过GSI执行Query操作，只读取符合条件的数据
    dynamic_frame = glueContext.create_dynamic_frame_from_options(
        connection_type="dynamodb",
        connection_options={
            "dynamodb.input.tableName": table_name,
            "dynamodb.input.indexName": "department-index",  # 替换为你的GSI名称
            "dynamodb.query.keyConditionExpression": "department = :val",
            "dynamodb.query.expressionAttributeValues": "{\":val\": {\"S\": \"Finance\"}}",
            "dynamodb.throughput.read.percent": "0.5"  # 按需调整，避免占用过多吞吐量
        }
    )
    df = dynamic_frame.toDF()
    return df

额外优化建议 #

• 减少不必要的列读取：如果不需要返回表中所有列，添加"dynamodb.projectionExpression": "col1, col2, department"参数，只读取业务所需列，进一步降低RCU消耗和数据传输量。
• 避免不必要的count操作：原代码中的df.count()会触发Spark的计算动作，浪费资源。使用Query操作时，若没有匹配数据，返回的DataFrame自然为空，直接返回即可。
• 调整吞吐量占比：根据业务需求调整dynamodb.throughput.read.percent，避免占用DynamoDB表的全部读写容量，影响其他业务。

备选方案：服务端过滤（Scan带Filter） #

如果暂时无法创建GSI，可使用DynamoDB的服务端过滤（仍为Scan操作，但只返回符合条件的数据），比客户端过滤更高效，但仍会扫描全表（消耗全表的RCU），仅作为临时过渡方案：

def fetch_data(table_name):
    dynamic_frame = glueContext.create_dynamic_frame_from_options(
        connection_type="dynamodb",
        connection_options={
            "dynamodb.input.tableName": table_name,
            "dynamodb.scan.filterExpression": "department = :val",
            "dynamodb.scan.expressionAttributeValues": "{\":val\": {\"S\": \"Finance\"}}",
            "dynamodb.throughput.read.percent": "0.5"
        }
    )
    df = dynamic_frame.toDF()
    return df

注：此方案仍会扫描全表，仅减少返回的数据量，RCU消耗和全表扫描一致，优先推荐Query+GSI方案。

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