AWS Glue每日运行月度报表仅聚合单日数据问题咨询

针对你的需求，核心思路是每日运行时动态计算当月日期范围，全量聚合当月1号至当天的数据，输出时覆盖当月的报表分区，具体实现步骤如下：

1. 动态计算日期参数 #

在Glue作业中先确定两个关键日期：

• credit_date：固定为当月第一天（比如2023-05-19运行时，值为2023-05-01）
• 数据过滤的结束日期：当前运行日期

用PySpark实现的代码示例：

from pyspark.sql.functions import current_date, date_trunc, col

# 生成当月第一天作为credit_date
credit_date = date_trunc('month', current_date()).cast('date')
# 获取当前运行日期作为数据截止日期
end_date = current_date()

2. 读取并过滤数据源 #

从S3读取Parquet格式的数据源，过滤出当月1号至当天的所有数据：

# 读取原始Parquet数据
raw_data_df = spark.read.parquet("s3://your-source-bucket/path-to-parquet-data/")

# 过滤日期范围内的数据（假设原始数据的日期字段为transaction_date）
filtered_df = raw_data_df.filter(col("transaction_date").between(credit_date, end_date))

3. 执行月度聚合逻辑 #

根据报表需求执行聚合操作，同时确保聚合结果的credit_date字段固定为当月第一天：

# 示例聚合逻辑：按维度分组，统计金额总和与交易数
aggregated_df = filtered_df.groupBy(credit_date.alias("credit_date"), "dimension_col1", "dimension_col2") \
                           .agg(
                               sum("transaction_amount").alias("total_amount"),
                               count("transaction_id").alias("transaction_count")
                           )

4. 输出并覆盖当月报表 #

将聚合结果写入S3时，采用覆盖模式，并按credit_date分区存储，这样每日运行都会替换当月的报表数据：

# 输出到目标S3路径，覆盖当月分区
aggregated_df.write.mode("overwrite") \
             .partitionBy("credit_date") \
             .parquet("s3://your-report-bucket/monthly-agg-report/")

5. Athena验证的配套处理 #

为了让Athena能及时获取最新的报表数据，需要同步分区信息：

• 方式一：在Glue任务执行完成后，调用Athena的分区修复命令：

import boto3

athena_client = boto3.client('athena')
# 替换为你的Athena数据库和表名
db_name = "your_athena_database"
table_name = "monthly_report_table"

# 执行分区修复
athena_client.start_query_execution(
    QueryString=f"MSCK REPAIR TABLE {db_name}.{table_name};",
    ResultConfiguration={'OutputLocation': 's3://your-athena-result-bucket/'}
)

• 方式二：配置Glue Crawler定期爬取报表存储路径，自动同步分区信息。

6. 调度设置 #

使用Glue触发器或CloudWatch Events，将任务设置为每日固定时间运行（比如凌晨），确保每日自动更新当月的聚合数据。

这种方案实现简单、数据一致性有保障，即使任务重复运行也不会产生脏数据（每次都会覆盖当月的全量聚合结果）。

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