Airflow最佳实践：多数据源同步至PostgreSQL数仓的技术问询

1. 在Airflow任务间共享数据是否为不良/错误做法？ #

不是绝对的错误，但属于需要谨慎使用的方式。Airflow的任务设计原则是无状态、独立运行，不同任务可能部署在不同Worker节点上，直接通过内存变量共享数据完全不可行。

如果是通过共享存储（比如NAS、对象存储）传递文件，是可行的，但要注意几个问题：

• 必须处理文件命名冲突、写入锁，避免多任务同时操作同一文件
• 要定期清理临时文件，防止存储资源耗尽
• 这种方式会增加任务间的依赖复杂度，排查问题时难度更高

如果是小体量数据，也可以用Airflow自带的XCom传递，但XCom默认有48KB的大小限制，不适合大数据量场景。

2. 是否存在其他更优的实现方案？ #

推荐**「数据落地中间存储+任务解耦」**的架构，完全替代任务间直接共享数据的模式：

• 把原「从数据源获取数据」的任务拆改为「从数据源拉取数据并写入中间存储」，中间存储可以选对象存储、临时PostgreSQL表、本地Parquet文件等
• 「插入数据仓库」的任务改为「从中间存储读取数据并写入目标仓库」

这种方案的优势：

• 任务完全独立，某一个数据源拉取失败不影响其他任务的重试或执行
• 中间存储的数据可复用，后续需要重新同步时无需再次调用API或读取源文件
• 便于添加数据校验、监控环节，比如新增任务检查中间存储的数据完整性

针对不同数据源还可以做针对性优化：

• PostgreSQL源数据：直接用SQL语句（INSERT INTO dw_table SELECT * FROM source_table WHERE ...）同步，或者用逻辑复制，减少中间落地步骤
• API数据源：如果支持增量拉取，每次拉取后记录最大时间戳或分页标记，存在Airflow变量或数据库中，下次拉取时直接复用

3. 若实现一个从源获取数据并插入至另一数据库的任务，该任务是否不具备幂等性？ #

不一定，完全取决于你的实现逻辑：

• 如果是全量覆盖写入（比如先TRUNCATE目标表再插入），重复执行结果一致，具备幂等性
• 如果是增量追加且无去重逻辑，重复执行会导致数据重复，不具备幂等性
• 要实现幂等性，可以做这些优化：
  • 针对数据库插入，使用INSERT ... ON CONFLICT (唯一键) DO NOTHING或INSERT ... ON CONFLICT (唯一键) UPDATE语句
  • 针对API拉取，每次拉取时带上时间范围或上次拉取的最后标记，只拉取新增数据
  • 针对Parquet文件，同步前校验文件的唯一标识（比如哈希值），避免重复处理同一文件

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