<p>We moved away from the Celery Executor in Airflow 1.10.0 because of some limitations of execution and right now we're using <code>KubernetesExecutor</code>. </p>

<p>Right now we're not able to parallelize all the tasks in some DAGs even when we change the <code>subdag_operator</code> in the code directly: <a href="https://github.com/apache/incubator-airflow/blob/v1-10-stable/airflow/operators/subdag_operator.py#L38" rel="nofollow noreferrer">https://github.com/apache/incubator-airflow/blob/v1-10-stable/airflow/operators/subdag_operator.py#L38</a></p>

<p>Our expectations it's that with these modifications and using Kubernetes Executors we can fan out the execution of all tasks at the same time but we have the same behavior of the <code>SequentialExecutor</code>.</p>

<p>This is the behavior that we have right now:</p>

<p><a href="https://i.stack.imgur.com/EX9Jb.png" rel="nofollow noreferrer"><img src="https://i.stack.imgur.com/EX9Jb.png" alt="enter image description here"></a></p>

<p>We would like to execute all of them at the same time using <code>KubernetesExecutor</code>.</p>

<p>Kubernetes Executor in Airflow will turn all the first level of tasks into a worker pod with Local Executor.</p>
<p>It means that you will get the Local Executor to execute your <code>SubDagOperator</code>.</p>
<p>In order to run the tasks under SubDagOperator after the spawning the worker pod, you will need to specify the configuration <code>parallelism</code> for the worker pod. So, in case you are using the YAML format for worker pod, you will need to edit it to something like this.</p>
<pre class="lang-yaml prettyprint-override"><code>apiVersion: v1
kind: Pod
metadata:
  name: dummy-name
spec:
  containers:
    - args: []
      command: []
      env:
        ###################################
        # This is the part you need to add
        ###################################
        - name: AIRFLOW__CORE__PARALLELISM
          value: 10
        ###################################
        - name: AIRFLOW__CORE__EXECUTOR
          value: LocalExecutor
        # Hard Coded Airflow Envs
        - name: AIRFLOW__CORE__FERNET_KEY
          valueFrom:
            secretKeyRef:
              name: RELEASE-NAME-fernet-key
              key: fernet-key
        - name: AIRFLOW__CORE__SQL_ALCHEMY_CONN
          valueFrom:
            secretKeyRef:
              name: RELEASE-NAME-airflow-metadata
              key: connection
        - name: AIRFLOW_CONN_AIRFLOW_DB
          valueFrom:
            secretKeyRef:
              name: RELEASE-NAME-airflow-metadata
              key: connection
      envFrom: []
      image: dummy_image
      imagePullPolicy: IfNotPresent
      name: base
      ports: []
      volumeMounts:
        - mountPath: &quot;/opt/airflow/logs&quot;
          name: airflow-logs
        - mountPath: /opt/airflow/dags
          name: airflow-dags
          readOnly: false
        - mountPath: /opt/airflow/dags
          name: airflow-dags
          readOnly: true
          subPath: repo/tests/dags
  hostNetwork: false
  restartPolicy: Never
  securityContext:
    runAsUser: 50000
  nodeSelector:
    {}
  affinity:
    {}
  tolerations:
    []
  serviceAccountName: 'RELEASE-NAME-worker-serviceaccount'
  volumes:
    - name: dags
      persistentVolumeClaim:
        claimName: RELEASE-NAME-dags
    - emptyDir: {}
      name: airflow-logs
    - configMap:
        name: RELEASE-NAME-airflow-config
      name: airflow-config
    - configMap:
        name: RELEASE-NAME-airflow-config
      name: airflow-local-settings
</code></pre>
<p>Then, <code>SubDagOperator</code> will follow the <code>parallelism</code> specified to run the tasks in parallel.</p>
