Issue 1:
Overview

Some Executors, currently a subset of the local Executors, run in a single threaded fashion and have certain limitations and requirements, many of which are hardcoded. To add a new single threaded Executor would require changes to core Airflow code.

Note: This coupling often shows up with SQLite compatibility checks since it does not support multiple connections.

Examples

2a) SQLite check done in configuration.py:

airflow/airflow/configuration.py

Lines 412 to 419
in
26f94c5

is_executor_without_sqlite_support = self.get("core", "executor") not in (

'DebugExecutor',

'SequentialExecutor',

)

is_sqlite = "sqlite" in self.get('database', 'sql_alchemy_conn')

if is_sqlite and is_executor_without_sqlite_support:

raise AirflowConfigException(f"error: cannot use sqlite with the {self.get('core', 'executor')}")

if is_sqlite:

2b) When running in standalone mode SQLite compatibility is checked:

airflow/airflow/cli/commands/standalone_command.py

Lines 160 to 165
in
26f94c5

if "sqlite" in conf.get("database", "sql_alchemy_conn"):

self.print_output("standalone", "Forcing executor to SequentialExecutor")

env["AIRFLOW__CORE__EXECUTOR"] = executor_constants.SEQUENTIAL_EXECUTOR

else:

self.print_output("standalone", "Forcing executor to LocalExecutor")

env["AIRFLOW__CORE__EXECUTOR"] = executor_constants.LOCAL_EXECUTOR

2c) Sensors in poke mode can block execution of DAGs when running with single process Executors, currently hardcoded to DebugExecutor (although should also include SequentialExecutor):

airflow/airflow/sensors/base.py

Line 243
in
27e2101

if conf.get('core', 'executor') == "DebugExecutor":

Proposal

A static method or attribute on the Executor class which can be checked by core code.

There is a precedent already set with the supports_ad_hoc_ti_run attribute, see:

airflow/airflow/executors/kubernetes_executor.py

Line 435
in
fb741fd

supports_ad_hoc_ti_run: bool = True

airflow/airflow/www/views.py

Lines 1735 to 1737
in
26f94c5

if not getattr(executor, "supports_ad_hoc_ti_run", False):

msg = "Only works with the Celery, CeleryKubernetes or Kubernetes executors"

return redirect_or_json(origin, msg, "error", 400)

Issue 2:
Description

The databricks provider can be updated to a deferrable task introduced in airflow 2.2.0. This will significnatly reduce cpu usage, memory usage on LocalExecutor and improve reliability since state is stored in the airflow metastore.

Use case/motivation

Currently the databricks operators works by calling the databricks REST API to submit a spark job and polling it every 20 secs to check when it is done. This creates the following inefficiencies:

If the airflow executor process crashes, a duplicate job can be created in databricks since airflow doesn't save the databricks job run id.

Each task instance of the databricks operators run its own process, however 95% of the time the process is idle and waiting to repoll the databricks API. If you want to 50 jobs in parallel, you will use a non trivial amount of memory / CPU just to poll a rest API.

Related issues

No response

Are you willing to submit a PR?

Yes I am willing to submit a PR!

Code of Conduct

I agree to follow this project's Code of Conduct