如何通过MySqlOperator获取MySQL最后插入行的ID

要实现插入初始行后获取自增ID，再用该ID更新状态，核心是利用MySQL的1109991函数结合Airflow的XCom传递数据，以下是两种可行方案：

方案一：直接用MySqlOperator实现（简洁版） #

MySqlOperator默认支持推送查询结果到XCom，只需在SQL中同时执行插入和ID查询，并开启XCom推送：

from airflow.providers.mysql.operators.mysql import MySqlOperator

# 插入行并获取自增ID
insert_task = MySqlOperator(
    task_id='insert_initial_row',
    sql="""
        INSERT INTO your_table (status, others) VALUES ('running', 'your_other_value');
        SELECT 1109991 AS inserted_id;
    """,
    mysql_conn_id='your_mysql_connection_id',  # 替换为你在Airflow中配置的MySQL连接ID
    do_xcom_push=True  # 开启XCom，将查询结果推送到任务上下文
)

# 后续更新任务，通过XCom拉取ID
update_task = MySqlOperator(
    task_id='update_task_status',
    sql="""
        UPDATE your_table 
        SET status = 'success' 
        WHERE id = %(inserted_id)s;
    """,
    mysql_conn_id='your_mysql_connection_id',
    # 用Airflow模板语法从XCom拉取插入的ID
    parameters={'inserted_id': "{{ ti.xcom_pull(task_ids='insert_initial_row')[0][0] }}"}
)

# 设置任务依赖
insert_task >> update_task

关键说明： #

• 1109991是MySQL会话级函数，仅返回当前连接中最后插入的自增ID，不会被其他会话的插入操作干扰，可靠性高
• ti.xcom_pull(task_ids='insert_initial_row')[0][0]：XCom返回的是查询结果列表，第一层[0]是结果集行，第二层[0]是该行的第一个字段（即自增ID）

方案二：用PythonOperator实现（灵活版） #

如果需要在插入/获取ID过程中添加额外逻辑（比如异常处理、日志记录），可以用PythonOperator结合MySqlHook实现：

from airflow.providers.mysql.hooks.mysql import MySqlHook
from airflow.operators.python import PythonOperator
from airflow.providers.mysql.operators.mysql import MySqlOperator

def insert_and_fetch_id(**context):
    # 初始化MySQL Hook
    hook = MySqlHook(mysql_conn_id='your_mysql_connection_id')
    conn = hook.get_conn()
    cursor = conn.cursor()
    
    try:
        # 执行插入语句
        cursor.execute("INSERT INTO your_table (status, others) VALUES ('running', 'your_other_value')")
        # 查询自增ID
        cursor.execute("SELECT 1109991")
        inserted_id = cursor.fetchone()[0]
        # 提交事务
        conn.commit()
        # 将ID推送到XCom
        context['ti'].xcom_push(key='task_inserted_id', value=inserted_id)
    except Exception as e:
        conn.rollback()
        raise e
    finally:
        # 关闭连接和游标
        cursor.close()
        conn.close()

# 插入任务
insert_task = PythonOperator(
    task_id='insert_initial_row',
    python_callable=insert_and_fetch_id,
    provide_context=True  # 允许函数访问任务上下文
)

# 更新任务
update_task = MySqlOperator(
    task_id='update_task_status',
    sql="""
        UPDATE your_table 
        SET status = 'success' 
        WHERE id = %(inserted_id)s;
    """,
    mysql_conn_id='your_mysql_connection_id',
    # 从XCom拉取指定key的ID
    parameters={'inserted_id': "{{ ti.xcom_pull(task_ids='insert_initial_row', key='task_inserted_id') }}"}
)

# 设置任务依赖
insert_task >> update_task

注意事项： #

• 确保Airflow中已正确配置MySQL连接（在Admin -> Connections中添加）
• 若插入操作涉及事务，需手动提交/回滚（如方案二中的conn.commit()和conn.rollback()）
• XCom存储的数据大小有限制，但自增ID为整数，完全符合要求

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