Is it possible to force a single task to "use new python interpreter"?

[pedro-cf]

• https://airflow.apache.org/docs/apache-airflow/stable/configurations-ref.html#execute-tasks-new-python-interpreter)

The AIRFLOW__CORE__EXECUTE_TASKS_NEW_PYTHON_INTERPRETER setting forces all tasks to be executed via forking of the parent process, spawning a new python process, slower than fork. And I would like to avoid affecting all tasks.

Is it possible enable this for single tasks that actually require it, so we don't have to affect the performance of all other dags/tasks?

In my case it's a mapped task that is causing problems... Using multiprocessing module or other forms of parallelism without this setting enabled causes the following error:

• AssertionError: daemonic processes are not allowed to have children

Full traceback:

[2024-07-23, 12:14:17 UTC] {taskinstance.py:2905} ERROR - Task failed with exception
Traceback (most recent call last):
  File "/home/airflow/.local/lib/python3.12/site-packages/airflow/models/taskinstance.py", line 465, in _execute_task
    result = _execute_callable(context=context, **execute_callable_kwargs)
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/airflow/.local/lib/python3.12/site-packages/airflow/models/taskinstance.py", line 432, in _execute_callable
    return execute_callable(context=context, **execute_callable_kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/airflow/.local/lib/python3.12/site-packages/airflow/models/baseoperator.py", line 401, in wrapper
    return func(self, *args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/airflow/.local/lib/python3.12/site-packages/airflow/decorators/base.py", line 265, in execute
    return_value = super().execute(context)
                   ^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/airflow/.local/lib/python3.12/site-packages/airflow/models/baseoperator.py", line 401, in wrapper
    return func(self, *args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/airflow/.local/lib/python3.12/site-packages/airflow/operators/python.py", line 235, in execute
    return_value = self.execute_callable()
                   ^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/airflow/.local/lib/python3.12/site-packages/airflow/operators/python.py", line 252, in execute_callable
    return self.python_callable(*self.op_args, **self.op_kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/opt/airflow/dags/tags_cats/tags_cats.py", line 209, in ingest_file_task
    campaign_id = ingest_file(s3_hook, postgis_hook, bucket_name, file_path, dag_run_id=ti.run_id, map_index=ti.map_index, params=params)
                  ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/opt/airflow/dags/tags_cats/tasks/process_file.py", line 174, in ingest_file
    campaign_id = process_zip(tmp_path, postgis_hook, dag_run_id, map_index, params)
                  ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/opt/airflow/dags/tags_cats/tasks/process_file.py", line 146, in process_zip
    ingest_csv_to_postgres(
  File "/opt/airflow/dags/extras/utils/csv_utils.py", line 115, in ingest_csv_to_postgres
    futures = [executor.submit(func, chunk) for chunk in dfs]
               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/concurrent/futures/process.py", line 831, in submit
    self._start_executor_manager_thread()
  File "/usr/local/lib/python3.12/concurrent/futures/process.py", line 770, in _start_executor_manager_thread
    self._launch_processes()
  File "/usr/local/lib/python3.12/concurrent/futures/process.py", line 797, in _launch_processes
    self._spawn_process()
  File "/usr/local/lib/python3.12/concurrent/futures/process.py", line 807, in _spawn_process
    p.start()
  File "/usr/local/lib/python3.12/multiprocessing/process.py", line 118, in start
    assert not _current_process._config.get('daemon'), \
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
AssertionError: daemonic processes are not allowed to have children

[pedro-cf]

Using a PythonVirtualEnvOperator forces airflow to use the environment's interpreter and fixes this problem, although I think it would be a good feature to be able to force a singular task to use a new interpreter every time it is executed.

[potiuk]

Feel free to propose a PR for that one.