airflow-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "Galak (JIRA)" <j...@apache.org>
Subject [jira] [Commented] (AIRFLOW-1814) Add op_args and op_kwargs in PythonOperator templated fields
Date Mon, 13 Nov 2017 19:17:00 GMT

    [ https://issues.apache.org/jira/browse/AIRFLOW-1814?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16250028#comment-16250028
] 

Galak commented on AIRFLOW-1814:
--------------------------------

Thank you for your answer.

I agree, I can do it programmatically in the consume_value callable function, but then I would
add a dependency upon airflow objects.

It's a question of separation of concerns:
All reusable / generic functions should not know about how to retrieve an xcom value; Moreover,
they could be called with different values, depending on the context: one could call it with
a static value, another with a job execution property, another one with a value coming from
a previous task instance... Then I would have to create a python wrapper function for every
new call context.
IMHO, "consume_value" should only receive a value to consume, and apply its business logic
to it.

> Add op_args and op_kwargs in PythonOperator templated fields
> ------------------------------------------------------------
>
>                 Key: AIRFLOW-1814
>                 URL: https://issues.apache.org/jira/browse/AIRFLOW-1814
>             Project: Apache Airflow
>          Issue Type: Wish
>          Components: operators
>    Affects Versions: Airflow 1.8, 1.8.0
>            Reporter: Galak
>            Priority: Minor
>
> *I'm wondering if "_op_args_" and "_op_kwargs_" PythonOperator parameters could be templated.*
> I have 2 different use cases where this change could help a lot:
> +1/ Provide some job execution information as a python callable argument:+
> let's explain it through a simple example:
> {code}
> simple_task = PythonOperator(
>     task_id='simple_task',
>     provide_context=True,
>     python_callable=extract_data,
>     op_args=[
> 	"my_db_connection_id"
> 	"select * from my_table"
> 	"/data/{dag.dag_id}/{ts}/my_export.csv"
>     ],
>     dag=dag
> )
> {code}
> "extract_data" python function seems to be simple here, but it could be anything re-usable
in multiple dags...
> +2/ Provide some XCom value as a python callable argument:+
> Let's say I a have a task which is retrieving or calculating a value, and then storing
it in an XCom for further use by other tasks:
> {code}
> value_producer_task = PythonOperator(
>     task_id='value_producer_task',
>     provide_context=True,
>     python_callable=produce_value,
>     op_args=[
> 	"my_db_connection_id",
> 	"some_other_static_parameter",
> 	"my_xcom_key"
>     ],
>     dag=dag
> )
> {code}
> Then I can just configure a PythonCallable task to use the produced value:
> {code}
> value_consumer_task = PythonOperator(
>     task_id='value_consumer_task',
>     provide_context=True,
>     python_callable=consume_value,
>     op_args=[
> 	"{{ task_instance.xcom_pull(task_ids=None, key='my_xcom_key') }}"
>     ],
>     dag=dag
> )
> {code}
> I quickly tried the following class:
> {code}
> from airflow.operators.python_operator import PythonOperator
> class MyPythonOperator(PythonOperator):
>     template_fields = PythonOperator.template_fields + ('op_args', 'op_kwargs')
> {code}
> and it worked like a charm.
> So could these 2 arguments be added to templated_fields? Or did I miss some major drawback
to this change?



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

Mime
View raw message