？

Question

は、私はこのような、simplehttpoperatorをトリガーにしたい： 気流trigger_dag test_trigger --conf '{ "名"： "何か"}' は

iがkwargsから[ 'dag_run' を使用してパラメータを受け入れるためにpythonoperatorのpython_callableを使用します] .confの、と私は[「dag_run」]を渡したい。simplehttpoperatorにconfに、どのように私はそれを行うことができますか？誰でも助けることができますか？タスク間通信のために

cc_ = {} 


def run_this_func(ds, **kwargs): 
    cc_ = kwargs['dag_run'].conf 
    logging.info(cc_) 
    return cc_ 

run_this = PythonOperator(
    task_id='run_this', 
    provide_context=True, 
    python_callable=run_this_func, 
    dag=dag) 

http_task = SimpleHttpOperator(
    task_id='http_task', 
    http_conn_id='test_http', 
    method='POST', 
    endpoint='/api/v1/function', 
    data=cc_, 
    headers={"Authorization": "Basic YWRtaW46MTIzNDU2", "Accept": "application/json, text/plain, */*"}, 
    response_check=lambda response: True if "10000" in response.content else False, 
    dag=dag) 

http_task.set_upstream(run_this) 

Answer 1

、あなたは、XCOMをチェックするhttps://airflow.incubator.apache.org/concepts.html#xcoms

をお勧めします***** ***** UPDATE
（おかげ詳細についてダニエル）以下 は、いくつかのコードです@Chengzhiと@Danielへ

http_task = SimpleHttpOperator(
    task_id='http_task', 
    http_conn_id='test_http', 
    method='POST', 
    endpoint='/api/v1/function', 
    data=json.loads("{{ task_instance.xcom_pull(task_ids='run_this', key='return_value') }}"), 
    headers={"Authorization": "Basic YWRtaW46MTIzNDU2", "Accept": "application/json, text/plain, */*"}, 
    response_check=lambda response: True if "10000" in response.content else False, 
    dag=dag) 

Answer 2

感謝を：あなたは、あなたのSimpleHttpOperatorにあなたはXCOMを経由して戻り値を取得し、それを試してみることができます。気流にデフォルトJinja2のバージョンは2.8.1であるとJinja2のはバージョン2.9までは「tojson」という名前の組み込みフィルタが含まれていないため は、最後に私は、Jinja2の/ filter.pyでカスタムフィルタ「tojson」を書きました。

def do_tojson(value): 
    value = json.JSONEncoder().encode(value) 
    return value 

dagファイルでは、コードは次のとおりです。できます。

def run_this_func(ds, **kwargs): 
    cc_ = kwargs['dag_run'].conf 
    return cc_ 

run_this = PythonOperator(
    task_id='run_this', 
    provide_context=True, 
    python_callable=run_this_func, 
    dag=dag) 

http_task = SimpleHttpOperator(
    task_id='http_task', 
    http_conn_id='test_http', 
    method='POST', 
    endpoint='/api/v1/task', 
    data="{{ task_instance.xcom_pull(task_ids='run_this') |tojson}}", 
    headers={"Authorization": "Basic YWRtaW46MTIzNDU2", "Accept": "application/json, text/plain, */*", 
      "Content-Type": "application/json"}, 
    response_check=lambda response: True if "10000" in response.content else False, 
    dag=dag) 

http_task.set_upstream(run_this)