Airflow环境下Spark写入PostgreSQL作业整点精准调度问题咨询

一、DAG配置调整（解决整点触发&单运行实例问题） #

你遇到的提前触发问题核心来源于两个错误配置：动态生成的start_date、未明确时区；单实例限制只需补充一个配置即可：

• 替换动态start_date为静态固定值：每次DAG解析时动态计算的now会导致调度基准时间不断偏移，是触发时间紊乱的首要原因，必须改成固定的历史日期，示例：

# 替换原动态start_date，根据你的实际上线时间调整即可，不要设置为未来时间
"start_date": datetime(2024, 1, 1, tzinfo=timezone.utc)

• 显式指定DAG时区：避免Airflow系统时区和你预期的调度时区不一致导致的时间偏差，国内场景可设置为Asia/Shanghai，若使用UTC调度可保留UTC时区。
• 新增catchup=False配置：关闭历史任务补跑，配合你已设置的max_active_runs=1，可严格保证同一时间只有一个DAG运行实例，不会生成多余任务。
• 可选将schedule_interval替换为cron表达式0 * * * *，和@hourly效果完全一致，更直观明确是每小时整点触发。

调整后完整DAG示例：

from datetime import datetime, timezone
default_args = {
    'owner': 'lns',
    'depends_on_past': True,
    "start_date": datetime(2024, 1, 1, tzinfo=timezone.utc), # 替换为你的上线时间
    'email_on_failure': True,
    'email_on_retry': True,
    'retries': 0
}

with DAG(
    dag_id='kafka_stream',
    default_args=default_args,
    schedule_interval='0 * * * *', # 等价于@hourly，明确整点触发
    max_active_runs=1,
    catchup=False, # 关闭补跑，避免多实例
    timezone='UTC' # 若用北京时间可改为Asia/Shanghai
) as dag:
    SparkSubmitOperator(
        task_id='batch_ingestion',
        conn_id='spark_default',
        application=f'/usr/local/spark/app/demo.py',
        application_args=["{{ execution_date.int_timestamp }}"], # 传入调度逻辑时间戳
        conf={"spark.master":spark_master},
        total_executor_cores=1,
        packages="org.apache.spark:spark-streaming-kafka-0-8_2.11:2.4.4",
        executor_cores=2,
        executor_memory='6g',
        name='data_processor',
        jars=postgres_driver_jar,
        driver_class_path=postgres_driver_jar
    )

二、Spark代码适配调整 #

原代码中使用datetime.now()获取当前时间的逻辑有缺陷：如果任务因为集群资源不足出现调度延迟，实际运行时间晚于整点，你计算出来的epoch值就会偏移到下一个整点，出现数据统计错误。建议直接使用Airflow传入的调度逻辑时间戳，不需要自行计算时间：

调整后Spark代码示例：

import sys
# 接收Airflow传入的整点时间戳，直接作为epoch使用
epoch = int(sys.argv[1])
print("epoch ", epoch)

df = spark.read \
.format("jdbc") \
.option("url", URL) \
.option("dbtable", "tagpool_with_tag_raw") \
.option("user", "tsdbadmin") \
.option("password", "cgqu5zy3i1") \
.option("driver", "org.postgresql.Driver") \
.load()

df.createOrReplaceTempView("tagpool_with_tag_raw")
query = spark.sql("select * from tagpool_with_tag_raw WHERE input_time = " + str(epoch))
query.show()
query.write \
    .format("jdbc") \
    .option("url", URL) \
    .option("dbtable", "tagpool_tag_raw") \
    .option("user", USER) \
    .option("password", PW) \
    .option("driver", DRIVER).save(mode='append')

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