mysql - 如何在 MySqlOperator 中使用 airflow xcoms

Question

def mysql_operator_test():
    DEFAULT_DATE = datetime(2017, 10, 9)
    t = MySqlOperator(
        task_id='basic_mysql',
        sql="SELECT count(*) from table 1 where id>100;",
        mysql_conn_id='mysql_default',
        dag=dag)
    t.run(start_date=DEFAULT_DATE, end_date=DEFAULT_DATE, ignore_ti_state=False)

run_this = PythonOperator(
    task_id='getRecoReq',
    python_callable=mysql_operator_test,
    # xcom_push=True,
    dag=dag)

task2 = PythonOperator(
    task_id= 'mysql_select',
    provide_context=True,
    python_callable = blah,
    templates_dict = {'requests': "{{ ti.xcom_pull(task_ids='getReq') }}" },
    dag=dag)

run_this.set_downstream(task2)

我想使用 xcoms 捕获 MySqlOperator 返回的计数。有人可以就此提供指导吗？

Answer 1

你非常接近!但是，您问这个问题的方式有点反模式。您不想在 Airflow 中跨任务共享数据。此外，您不想像在 mysql_operator_test 中那样使用运算符。很诱人，我刚开始的时候也做过同样的事情。

我尝试了与此非常相似但使用 SFTP 连接的方法。我最终只是在 PythonOperator 中完成了所有操作并使用了底层 Hook 。

我建议您在 python_callable 中使用 MySQLHook。像这样:

def count_mysql_and_then_use_the_count():
    """
    Returns an SFTP connection created using the SSHHook
    """
    mysql_hook = MySQLHook(...)
    cur = conn.cursor()
    cur.execute("""SELECT count(*) from table 1 where id>100""")
    for count in cur: 
       # Do something with the count...

我不确定这是否会按原样工作，但我的想法是在你的 Python 可调用函数中使用一个钩子(Hook)，我不经常使用 MySQLHook 但我使用 SSHHook 完成此操作并且效果很好。