如何在Airflow中编写执行MySQL查询的任务

在Airflow中编写执行MySQL查询的任务可以通过使用PythonOperator来执行查询的Python函数。以下是一个简单的示例：

from airflow import DAGfrom airflow.operators.python_operator import PythonOperatorfrom airflow.operators.mysql_operator import MySqlOperatorfrom datetime import datetimedefault_args = {    'owner': 'airflow',    'depends_on_past': False,    'start_date': datetime(2021, 1, 1),    'email_on_failure': False,    'email_on_retry': False,    'retries': 1}dag = DAG('mysql_query_dag', default_args=default_args, schedule_interval='@daily')def execute_mysql_query():    # 连接MySQL数据库    conn = MySQLdb.connect(host="localhost", user="root", passwd="password", db="database")    cursor = conn.cursor()        # 执行查询    cursor.execute("SELECT * FROM table")        # 获取结果    rows = cursor.fetchall()    for row in rows:        print(row)        # 关闭连接    conn.close()mysql_task = PythonOperator(    task_id='execute_mysql_query',    python_callable=execute_mysql_query,    dag=dag)mysql_task

在这个例子中，首先创建了一个DAG，并定义了一个Python函数execute_mysql_query，该函数连接到MySQL数据库，执行查询并打印结果。然后使用PythonOperator来执行这个函数，并将其添加到DAG中。当DAG运行时，该任务将连接到MySQL数据库执行查询。