如何自动调度同步S3新增CSV数据至Athena的CTAS表并更新报表

一、前提准备：确保外部表能识别新增数据 #

• 给S3的CSV文件设置可识别的路径规则，比如按日期分区存储：s3://your-bucket/csv-data/date=2024-05-20/，或者文件名包含时间戳（如data_20240520.csv）。
• 如果数据本身带业务时间字段（比如create_time），确保外部表已包含该字段，用于后续过滤新增数据。

二、选择适合的同步方案 #

方案1：增量计算+原子替换（适合中小数据量） #

每次仅处理新增数据，合并到现有CTAS表，避免全量重算：

1. 创建临时增量表：过滤外部表中未同步到CTAS表的数据：

   CREATE TABLE temp_increment_data
   WITH (format = 'PARQUET')
   AS
   SELECT * FROM external_csv_table
   -- 按业务时间过滤，取CTAS表中未存在的新数据
   WHERE create_time > (SELECT COALESCE(MAX(update_time), '1970-01-01') FROM your_ctas_table)
   -- 也可按S3路径过滤，比如仅取当日新增文件
   -- AND "$path" LIKE '%2024-05-20%'
   

2. 对增量数据执行聚合去重：复用原CTAS表的计算逻辑：

   CREATE TABLE temp_agg_increment
   WITH (format = 'PARQUET')
   AS
   SELECT user_id, date, SUM(amount) AS total_amount, COUNT(*) AS order_count
   FROM temp_increment_data
   GROUP BY user_id, date
   -- 加入原有的去重逻辑，比如用ROW_NUMBER()筛选唯一记录
   

3. 合并数据并原子替换原CTAS表：避免报表查询时出现数据不一致：

   -- 创建新的全量结果表
   CREATE TABLE new_ctas_table
   WITH (format = 'PARQUET')
   AS
   SELECT * FROM your_ctas_table
   UNION ALL
   SELECT * FROM temp_agg_increment
   -- 再次去重/聚合，确保无重复数据
   GROUP BY user_id, date, total_amount, order_count
   

   执行原子替换：

   ALTER TABLE your_ctas_table SWAP WITH new_ctas_table;
   

   最后清理临时表：

   DROP TABLE temp_increment_data;
   DROP TABLE temp_agg_increment;
   DROP TABLE new_ctas_table; -- 替换后原new表会变为旧的your_ctas_table，按需清理
   

方案2：分区表+增量同步（适合大数据量） #

通过分区减少扫描数据量，提升同步效率：

1. 将外部表改为分区表：用AWS Glue Crawler扫描S3分区路径（如date=yyyy-mm-dd）自动创建分区，或手动添加：

   ALTER TABLE external_csv_table ADD PARTITION (date='2024-05-20') LOCATION 's3://your-bucket/csv-data/date=2024-05-20/';
   

2. CTAS表同步设置分区：初始创建时就按分区构建：

   CREATE TABLE your_ctas_table
   WITH (format = 'PARQUET', partitioned_by = ARRAY['date'])
   AS
   SELECT user_id, date, SUM(amount) AS total_amount
   FROM external_csv_table
   GROUP BY user_id, date;
   

3. 同步新增分区数据：仅处理S3新增的对应分区：

   -- 刷新外部表分区（Glue Crawler可自动完成此步骤）
   MSCK REPAIR TABLE external_csv_table;
   -- 将新增分区的聚合数据插入CTAS表
   INSERT INTO your_ctas_table
   SELECT user_id, date, SUM(amount) AS total_amount
   FROM external_csv_table
   WHERE date = '2024-05-20' -- 替换为新增的日期
   GROUP BY user_id, date;
   

三、自动调度实现 #

用AWS服务组合完成自动同步：

• 定时调度：通过EventBridge设置定时规则（比如每天凌晨2点），触发Lambda函数或Glue Job执行同步SQL。Lambda可调用Athena的StartQueryExecution API执行逻辑；Glue Job更适合大数据量的脚本处理。
• 事件触发：给S3桶设置PutObject事件通知，触发Lambda函数识别新增文件的日期/路径，自动执行对应分区的同步逻辑。

四、关键注意事项 #

• 原子性保障：必须用ALTER TABLE SWAP或先建新表再替换的方式，避免报表查询到不完整数据。
• 去重逻辑复用：新增数据可能包含重复记录，同步时要保留原有的去重/聚合规则，确保数据一致性。
• 成本控制：增量同步能大幅减少Athena的扫描数据量，降低计费成本。
• 监控告警：给Athena查询、Lambda/Glue Job配置CloudWatch告警，及时发现同步失败问题。

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