Airflow LocalExecutor - please provide example to execute DAG in local executor

[hemaroh]

Hello,

I have made Executor to LocalExecuter in congratulation file and ran the dag. But still DAG is failed by saying event logs as "Executor CeleryExecutor(parallelism=64) <TaskInstance queue and failed.

Please help with code where we can provide executor in DAG code itself ?

Airflow version - 2.10.2
Redis, airflow-celery, celery packages installed with latest version.

Thanks
Nav

[potiuk]

I think you would increase your chance of getting help if you explain better - possibly with examples - what you try to do, because so far I think no-one knows what you tried to do and how you tried to do it.

[hemaroh]

Executor = LocalExecutor

Made local executor in airflow.cfg.

Code:
from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta
import time
def hello_function():
print('Hello, this is the first task of the DAG')
time.sleep(5)
def last_function():
print('DAG run is done.')
def sleeping_function():
print("Sleeping for 5 seconds")
time.sleep(5)
with DAG(
dag_id="local_executor_demo",
start_date=datetime(2021,1,1),
schedule_interval="@hourly",
catchup=False) as dag:
task1=PythonOperator(
task_id="hello_function",
python_callable=hello_function
)
task2_1=PythonOperator(
task_id="sleepy_1",
python_callable=sleeping_function
)
task2_2=PythonOperator(
task_id="sleepy_2",
python_callable=sleeping_function
)
task3=PythonOperator(
task_id="bye_function",
python_callable=last_function
)
task1>>[task2_1,task2_2]>>task3

[hemaroh]

Executor = LocalExecutor

Made local executor in airflow.cfg.

Code:
from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta
import time
def hello_function():
print('Hello, this is the first task of the DAG')
time.sleep(5)
def last_function():
print('DAG run is done.')
def sleeping_function():
print("Sleeping for 5 seconds")
time.sleep(5)
with DAG(
dag_id="local_executor_demo",
start_date=datetime(2021,1,1),
schedule_interval="@hourly",
catchup=False) as dag:
task1=PythonOperator(
task_id="hello_function",
python_callable=hello_function
)
task2_1=PythonOperator(
task_id="sleepy_1",
python_callable=sleeping_function
)
task2_2=PythonOperator(
task_id="sleepy_2",
python_callable=sleeping_function
)
task3=PythonOperator(
task_id="bye_function",
python_callable=last_function
)
task1>>[task2_1,task2_2]>>task3

=====