如何在MySqlOperator中使用气流xcoms [英] How to use airflow xcoms with MySqlOperator

问题描述

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返回的计数.有人可以指导一下吗?

I want to capture the count returned by the MySqlOperator using xcoms. Can someone please guide regarding the same?

推荐答案

您非常亲密！但是，您问这个问题的方式有点像反模式.您不想在Airflow中的各个任务之间共享数据.另外，您也不想像在mysql_operator_test中那样使用运算符.这很诱人，刚开始时我做了同样的事情.

You're very close! However, the way you're asking this question is kind of an anti-pattern. You don't want to share data across tasks in Airflow. Also, you don't want to use the operator like you are in mysql_operator_test. It's tempting, I did the same thing when I was getting started.

我尝试了与此非常相似的东西，但是使用了SFTP连接.我最终只是在PythonOperator内完成了所有操作，并使用了底层的钩子.

I tried something very similar to this but with SFTP connections. I eventually just did everything inside of the PythonOperator and used the underlying hooks.

我建议您在python_callable内使用MySQLHook.像这样:

I'd recommend you use the MySQLHook inside of a python_callable. Something like this:

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内使用钩子，我不经常使用MySQLHook，但是我使用SSHHook效果很好.

I'm not sure if this will work as is but the idea is use a hook inside your Python callable, I don't use the MySQLHook often but I did this with the SSHHook and it's been working great.

这篇关于如何在MySqlOperator中使用气流xcoms的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！