Airflow expand functionality in DatabricksSubmitRunOperator

[nikhil-koul]

Hi All,

I'm trying to implement Airflow expand functionality in DatabricksSubmitRunOperator.
If you, please provide me with examples of how it is done in DatabricksSubmitRunOperator?

Also, by using expand function, the tasks will be in parallel execution or sequential.

Regards,
Nikhil

[potiuk]

You should look at exisitng examples in https://airflow.apache.org/docs/apache-airflow/stable/concepts/dynamic-task-mapping.html?highlight=expand#assigning-multiple-parameters-to-a-non-taskflow-operator and apply it to the operator you want to use. Airlfow has > 3000 operators and it is quite unwise to expect specific examples for single feature for a specific operator.

Also read the Dynamic Tasks mapping docs to understand how it works - each of the dynamic tasks runs as separate task, so depending on your tasks run in parallel but they might run either sequentially if you wish by using pools, queues and other mechanisms available. Just treat dynamic tasks as if you had all the dyamic tasks defined individually in your DAG.

[sjain-cn]

Astronomer Runtime 6.0.2 based on Airflow 2.4.1+astro.1 . It is working with PythonOperator.

[potiuk]

Did you follow the example and used partial().expand() chain? I think you tried to use expand directly on the class and this is not the way it works.

[potiuk]

And I am really curious how it worked on the PythonOperator ? You probably made some bad assumption or mistake, or maybe you have somewhat modified PythonOperator:

>>> PythonOperator.expand([])
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
AttributeError: type object 'PythonOperator' has no attribute 'expand'

[sjain-cn]

Sample DAG-

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime
from airflow.decorators import task

def helloWorld():
print("hello World!")

@task
def get_market_list():
return ['germany','france']

@task
def helloWorld1(market: str):
print("helloWorld1"+" "+ market)

with DAG(dag_id="hello_world_dag",
start_date=datetime(2021,1,1),
schedule_interval="@hourly",
catchup=False) as dag:

    helloWorld1.expand(market=get_market_list())

[potiuk]

Ths is not a PythonOperator. It's a decorated/TaskFlow operator and they are completely differently treated (the docs explain all that) and the chapter I refered to as "Mapping with non-TaskFlow operators".