AWS Glue作业如何筛选DynamoDB表中time字段对应前一天的数据

核心修改逻辑 #

你需要在字段类型转换完成后新增日期过滤步骤，具体修改如下：

1. 导入Spark内置日期处理函数，用于计算前一天日期和匹配日期字段
2. 在ApplyMapping转换后新增过滤规则，仅保留time字段日期为前一天的记录
3. 将过滤后的结果输出到S3

修改后完整代码 #

import sys
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from pyspark.sql.functions import to_date, date_sub, current_date
from awsglue.context import GlueContext
from awsglue.job import Job
import re

args = getResolvedOptions(sys.argv, ['JOB_NAME'])

sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args['JOB_NAME'], args)

DataSource0 = glueContext.create_dynamic_frame.from_catalog(database = "sampledb", table_name = "tablex", transformation_ctx = "DataSource0")

Transform1 = ApplyMapping.apply(frame = DataSource0, mappings = [("allresults.paymentCapabilityResult.paymentCapabilityCheckResult", "boolean", "allresults.paymentCapabilityResult.paymentCapabilityCheckResult", "boolean"), ("time", "string", "time", "timestamp")], transformation_ctx = "Transform1")

# 新增过滤步骤：仅保留time为前一天的数据
Filter2 = Filter.apply(
    frame = Transform1,
    f = lambda row: to_date(row["time"]) == date_sub(current_date(), 1),
    transformation_ctx = "Filter2"
)

DataSink0 = glueContext.write_dynamic_frame.from_options(frame = Filter2, connection_type = "s3", format = "json", connection_options = {"path": "s3://xxxxx/output/", "partitionKeys": []}, transformation_ctx = "DataSink0")
job.commit()

注意事项 #

• 上述代码默认使用UTC时区计算日期，如果你的time字段使用其他时区，可在to_date函数中指定时区参数，例如东八区使用to_date(row["time"], "UTC+8")
• 如果DynamoDB表数据量极大，可将过滤条件下推到DynamoDB扫描阶段，进一步降低数据读取量：在create_dynamic_frame.from_catalog方法的additional_options参数中添加DynamoDB过滤表达式

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