### 在Airflow中基于MongoDB交易日实现自定义调度逻辑 在Airflow中，可以通过编写自定义的调度逻辑来实现基于MongoDB中的交易日进行调度。以下是具体方法和步骤的详细说明： #### 1. 自定义调度逻辑 Airflow默认使用`@dag`装饰器或通过`schedule_interval`参数设置调度规则，但这些规则通常是基于时间（如cron表达式）或外部触发器。为了实现基于MongoDB交易日的调度，需要结合PythonOperator或Sensor Operator动态获取MongoDB中的数据并决定是否触发任务。 #### 2. 使用PythonOperator动态调度 可以编写一个Python函数，连接到MongoDB并查询最新的交易日信息。如果满足条件（例如有新的交易日），则触发后续的任务。 ```python from airflow import DAG from airflow.operators.python import PythonOperator from datetime import datetime, timedelta import pymongo # 定义DAG default_args = { 'owner': 'airflow', 'depends_on_past': False, 'email_on_failure': False, 'email_on_retry': False, 'retries': 1, 'retry_delay': timedelta(minutes=5), } with DAG( 'mongo_based_scheduler', default_args=default_args, start_date=datetime(2023, 1, 1), schedule_interval=None, # 手动触发或通过API触发 catchup=False, ) as dag: def check_mongo_transaction_day(): """检查MongoDB中的最新交易日""" client = pymongo.MongoClient("mongodb://localhost:27017") db = client['transaction_db'] collection = db['transaction_days'] # 获取最新的交易日 latest_transaction_day = collection.find_one(sort=[("date", pymongo.DESCENDING)]) if latest_transaction_day and latest_transaction_day['date'] > datetime.now().date() - timedelta(days=1): print(f"New transaction day detected: {latest_transaction_day['date']}") return True else: print("No new transaction day detected.") return False def process_data(**kwargs): """处理数据的逻辑""" print("Processing data based on the latest transaction day.") # 检查交易日的任务 check_task = PythonOperator( task_id='check_transaction_day', python_callable=check_mongo_transaction_day, ) # 处理数据的任务 process_task = PythonOperator( task_id='process_data', python_callable=process_data, ) # 设置依赖关系 check_task >> process_task ``` 上述代码中，`check_mongo_transaction_day`函数会连接到MongoDB，并查询最新的交易日。如果发现新的交易日，则触发`process_data`任务[^1]。 #### 3. 使用Sensor等待特定条件 如果需要更灵活的方式，可以使用Airflow的Sensor Operator，例如`BaseSensorOperator`，来等待MongoDB中的交易日更新。 ```python from airflow.sensors.base import BaseSensorOperator class MongoTransactionDaySensor(BaseSensorOperator): def __init__(self, mongo_uri, db_name, collection_name, *args, **kwargs): super().__init__(*args, **kwargs) self.mongo_uri = mongo_uri self.db_name = db_name self.collection_name = collection_name def poke(self, context): client = pymongo.MongoClient(self.mongo_uri) db = client[self.db_name] collection = db[self.collection_name] latest_transaction_day = collection.find_one(sort=[("date", pymongo.DESCENDING)]) if latest_transaction_day and latest_transaction_day['date'] > datetime.now().date() - timedelta(days=1): self.log.info(f"New transaction day detected: {latest_transaction_day['date']}") return True else: self.log.info("No new transaction day detected yet.") return False # 在DAG中使用Sensor sensor_task = MongoTransactionDaySensor( task_id='wait_for_transaction_day', mongo_uri="mongodb://localhost:27017", db_name='transaction_db', collection_name='transaction_days', timeout=60*60, # 等待1小时 mode="reschedule", # 避免占用Worker资源 ) ``` 此传感器会在每次执行时检查MongoDB中的最新交易日，只有当检测到新交易日时才会继续执行下游任务[^5]。 #### 4. 元数据库与外部数据源结合 Airflow的元数据库通常用于存储任务运行状态等信息，而外部数据源（如MongoDB）可以作为业务逻辑的一部分参与调度决策。通过上述方式，可以将MongoDB中的交易日与Airflow的任务调度无缝结合[^3]。 ---