此问题的可能原因是，在Task 1中提取的记录格式与Task 2中预期的不匹配。解决此问题的一种方法是使用XCom，这是Airflow用于在任务之间共享信息的功能。

在Task 1中，可以使用XCom将记录推送到Task 2，如下所示：

from airflow.operators.python_operator import PythonOperator from airflow.models import XCom

def push_records(**context): # 提取记录 records = get_records() # 推送记录到XCom context['ti'].xcom_push(key='my_records', value=records)

task_1 = PythonOperator( task_id='task_1', python_callable=push_records, provide_context=True, dag=dag )

在Task 2中，可以访问Task 1中推送的记录，如下所示：

def pull_records(**context): # 从XCom中拉取记录 records = context['ti'].xcom_pull(key='my_records') # 使用记录 process_records(records)

task_2 = PythonOperator( task_id='task_2', python_callable=pull_records, provide_context=True, dag=dag )

这些代码片段需要合并到Airflow DAG中。这将确保Task 2能够获取Task 1推送的记录。