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) 

मैं गिनती MySqlOperator xcoms का उपयोग करके लौटे कैप्चर करना चाहते हैं के साथ हवा का प्रवाह xcoms उपयोग करने के लिए। क्या कोई इसके बारे में मार्गदर्शन कर सकता है?

Answer 1

आप बहुत करीब हैं! हालांकि, जिस तरह से आप इस सवाल पूछ रहे हैं वह एक विरोधी पैटर्न है। आप एयरफ्लो में कार्यों में डेटा साझा नहीं करना चाहते हैं। साथ ही, आप ऑपरेटर का उपयोग नहीं करना चाहते हैं जैसे कि आप mysql_operator_test में हैं। यह मोहक है, मैंने वही काम किया जब मैं शुरू कर रहा था।

मैंने कुछ ऐसा करने की कोशिश की लेकिन एसएफटीपी कनेक्शन के साथ। अंततः मैंने PythonOperator के अंदर सबकुछ किया और अंतर्निहित हुक का उपयोग किया।

मैं आपको MySQLHookpython_callable के अंदर उपयोग करने की सलाह दूंगा। कुछ इस तरह:

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... 

मुझे यकीन है कि अगर यह इसी रूप में काम करेंगे नहीं कर रहा हूँ लेकिन यह विचार अपने अजगर के अंदर एक हुक का उपयोग प्रतिदेय, मैं MySQLHook अक्सर उपयोग नहीं करते है, लेकिन मैं के साथ ऐसा किया SSHHook और यह बहुत अच्छा काम कर रहा है।