Airflow 3.1.1：LocalExecutor无限并行模式配置报错求助

Airflow官方文档提到LocalExecutor支持「无限并行」，但未明确具体配置方式。若直接将core.parallelism配置项设为0（文档说明0代表无限并行），会触发以下报错：

Traceback (most recent call last):
  File "/home/airflow/.local/bin/airflow", line 7, in <module>
    sys.exit(main())
             ^^^^^^
  File "/home/airflow/.local/lib/python3.12/site-packages/airflow/__main__.py", line 55, in main
    args.func(args)
  File "/home/airflow/.local/lib/python3.12/site-packages/airflow/cli/cli_config.py", line 49, in command
    return func(*args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^
  File "/home/airflow/.local/lib/python3.12/site-packages/airflow/utils/cli.py", line 114, in wrapper
    return f(*args, **kwargs)
           ^^^^^^^^^^^^^^^^^^
  File "/home/airflow/.local/lib/python3.12/site-packages/airflow/utils/providers_configuration_loader.py", line 54, in wrapped_function
    return func(*args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^
  File "/home/airflow/.local/lib/python3.12/site-packages/airflow/cli/commands/scheduler_command.py", line 52, in scheduler
    run_command_with_daemon_option(
  File "/home/airflow/.local/lib/python3.12/site-packages/airflow/cli/commands/daemon_utils.py", line 86, in run_command_with_daemon_option
    callback()
  File "/home/airflow/.local/lib/python3.12/site-packages/airflow/cli/commands/scheduler_command.py", line 55, in <lambda>
    callback=lambda: _run_scheduler_job(args),
                     ^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/airflow/.local/lib/python3.12/site-packages/airflow/cli/commands/scheduler_command.py", line 43, in _run_scheduler_job
    run_job(job=job_runner.job, execute_callable=job_runner._execute)
  File "/home/airflow/.local/lib/python3.12/site-packages/airflow/utils/session.py", line 100, in wrapper
    return func(*args, session=session, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/airflow/.local/lib/python3.12/site-packages/airflow/jobs/job.py", line 368, in run_job
    return execute_job(job, execute_callable=execute_callable)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/airflow/.local/lib/python3.12/site-packages/airflow/jobs/job.py", line 397, in execute_job
    ret = execute_callable()
          ^^^^^^^^^^^^^^^^^^
  File "/home/airflow/.local/lib/python3.12/site-packages/airflow/jobs/scheduler_job_runner.py", line 1085, in _execute
    for executor in self.job.executors:
                    ^^^^^^^^^^^^^^^^^^
  File "/usr/python/lib/python3.12/functools.py", line 998, in __get__
    val = self.func(instance)
          ^^^^^^^^^^^^^^^^^^^
  File "/home/airflow/.local/lib/python3.12/site-packages/airflow/jobs/job.py", line 164, in executors
    return ExecutorLoader.init_executors()
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/airflow/.local/lib/python3.12/site-packages/airflow/executors/executor_loader.py", line 222, in init_executors
    loaded_executor = cls.load_executor(executor_name)
                      ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/airflow/.local/lib/python3.12/site-packages/airflow/executors/executor_loader.py", line 273, in load_executor
    executor = executor_cls()
               ^^^^^^^^^^^^^^
  File "/home/airflow/.local/lib/python3.12/site-packages/airflow/executors/local_executor.py", line 154, in __init__
    super().__init__(parallelism=parallelism)
  File "/home/airflow/.local/lib/python3.12/site-packages/airflow/executors/base_executor.py", line 178, in __init__
    raise ValueError("parallelism is set to 0 or lower")
ValueError: parallelism is set to 0 or lower

解决方案 #

文档中提到的「无限并行」是指LocalExecutor本身没有内置的并行数量上限（区别于SequentialExecutor的串行执行），但Airflow的BaseExecutor会强制校验parallelism值必须大于0。要实现近似无限的并行效果，只需将core.parallelism设为一个足够大的正整数，具体步骤如下：

1. 打开Airflow配置文件（通常为airflow.cfg）
2. 确保executor配置为LocalExecutor：

[core]
executor = LocalExecutor

3. 将parallelism设为一个足够大的正整数（比如1000，数值可根据系统资源调整）：

[core]
parallelism = 1000

4. 重启Airflow调度器和Web服务，使配置生效

此时，LocalExecutor会在系统CPU、内存等资源允许的前提下，尽可能多地并行执行任务，达到文档所述的「无限并行」效果。

内容的提问来源于stack exchange，提问作者Alkis Mavridis