<h3 dir="auto">Apache Airflow version</h3>
<p dir="auto">2.3.2</p>
<h3 dir="auto">What happened</h3>
<p dir="auto">I am using the <code class="notranslate">KubernetesPodOperator</code> and I set the <code class="notranslate">resources</code> keyword argument in the <code class="notranslate">partial</code> function so that I can use dynamic task mapping. This doesn't work and breaks the code. The following is the errors It is giving :</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="Broken DAG: [/dags/foreach-mapper-fail.py] Traceback (most recent call last):
  File &quot;/home/airflow/.local/lib/python3.7/site-packages/airflow/models/baseoperator.py&quot;, line 287, in partial
    partial_kwargs[&quot;resources&quot;] = coerce_resources(partial_kwargs[&quot;resources&quot;])
  File &quot;/home/airflow/.local/lib/python3.7/site-packages/airflow/models/baseoperator.py&quot;, line 133, in coerce_resources
    return Resources(**resources)
TypeError: type object argument after ** must be a mapping, not V1ResourceRequirements"><pre class="notranslate"><code class="notranslate">Broken DAG: [/dags/foreach-mapper-fail.py] Traceback (most recent call last):
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/models/baseoperator.py", line 287, in partial
    partial_kwargs["resources"] = coerce_resources(partial_kwargs["resources"])
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/models/baseoperator.py", line 133, in coerce_resources
    return Resources(**resources)
TypeError: type object argument after ** must be a mapping, not V1ResourceRequirements
</code></pre></div>
<h3 dir="auto">What you think should happen instead</h3>
<p dir="auto">I suspect it is because <code class="notranslate">BaseOperator</code> is explicitly given <code class="notranslate">None</code> in the <code class="notranslate">resources</code> <a href="https://github.com/apache/airflow/blob/8a34d25049a060a035d4db4a49cd4a0d0b07fb0b/airflow/providers/cncf/kubernetes/operators/kubernetes_pod.py#L217">argument</a></p>
<h3 dir="auto">How to reproduce</h3>
<p dir="auto">Here is an example DAG to reproduce. Change the <code class="notranslate">namespace</code> in the <code class="notranslate">KubernetesPodOperator</code>:</p>
<div class="highlight highlight-source-python notranslate position-relative overflow-auto" dir="auto" data-snippet-clipboard-copy-content="from datetime import timedelta
from airflow import DAG
from airflow.contrib.operators.kubernetes_pod_operator import KubernetesPodOperator
from airflow.operators.bash import BashOperator
from airflow.utils.dates import days_ago
from kubernetes import client
from airflow.models.param import Param

default_args = {
    &quot;owner&quot;: &quot;Airflow&quot;,
    &quot;depends_on_past&quot;: False,
    &quot;start_date&quot;: days_ago(0),
    &quot;email&quot;: [&quot;airflow@example.com&quot;],
    &quot;email_on_failure&quot;: False,
    &quot;email_on_retry&quot;: False,
    &quot;retries&quot;: 0,
    &quot;retry_delay&quot;: timedelta(minutes=5),
}

dag = DAG(
    &quot;MapperK8sOperator-3&quot;,
    default_args=default_args,
    schedule_interval=timedelta(minutes=30),
    max_active_runs=1,
    user_defined_filters={&quot;to_list&quot;: lambda x: list(x)},
    concurrency=10,
    params={
        &quot;param1&quot;: Param(10, type=&quot;integer&quot;, minimum=0, maximum=20),
        &quot;param2&quot;: &quot;value2&quot;,
    },
)

# Generate 2 tasks
tasks = [&quot;task{}&quot;.format(i) for i in range(1, 3)]
example_dag_complete_node = BashOperator(
    task_id=&quot;example_dag_complete&quot;,
    dag=dag,
    bash_command=&quot;echo {{task_instance.xcom_pull(task_ids='task1',key='pod_name') | to_list }}&quot;,
)

org_dags = []


def make_cmds():
    return [
        [
            &quot;mkdir -p /airflow/xcom/;echo '[1,2,3,4,6]' &gt; /airflow/xcom/return.json;&quot;
        ],
        [
            &quot;mkdir -p /airflow/xcom/;echo '[1,2,3,4,109]' &gt; /airflow/xcom/return.json;&quot;
        ],
    ]


for task in tasks:

    bash_command = &quot;echo HELLO&quot;

    org_node = KubernetesPodOperator.partial(
        namespace=&quot;my-namespace&quot;,
        image=&quot;python&quot;,
        labels={&quot;foo&quot;: &quot;bar&quot;},
        image_pull_policy=&quot;Always&quot;,
        name=task,
        task_id=task,
        is_delete_operator_pod=True,
        resources = client.V1ResourceRequirements(requests={
            &quot;cpu&quot;: 1,
            &quot;memory&quot;: &quot;256M&quot;,
        }),
        do_xcom_push=True,
        get_logs=True,
        dag=dag,
    ).expand(cmds=make_cmds())

    org_node.set_downstream(example_dag_complete_node)
    org_node &gt;&gt; example_dag_complete_node
"><pre class="notranslate"><span class="pl-k">from</span> <span class="pl-s1">datetime</span> <span class="pl-k">import</span> <span class="pl-s1">timedelta</span>
<span class="pl-k">from</span> <span class="pl-s1">airflow</span> <span class="pl-k">import</span> <span class="pl-v">DAG</span>
<span class="pl-k">from</span> <span class="pl-s1">airflow</span>.<span class="pl-s1">contrib</span>.<span class="pl-s1">operators</span>.<span class="pl-s1">kubernetes_pod_operator</span> <span class="pl-k">import</span> <span class="pl-v">KubernetesPodOperator</span>
<span class="pl-k">from</span> <span class="pl-s1">airflow</span>.<span class="pl-s1">operators</span>.<span class="pl-s1">bash</span> <span class="pl-k">import</span> <span class="pl-v">BashOperator</span>
<span class="pl-k">from</span> <span class="pl-s1">airflow</span>.<span class="pl-s1">utils</span>.<span class="pl-s1">dates</span> <span class="pl-k">import</span> <span class="pl-s1">days_ago</span>
<span class="pl-k">from</span> <span class="pl-s1">kubernetes</span> <span class="pl-k">import</span> <span class="pl-s1">client</span>
<span class="pl-k">from</span> <span class="pl-s1">airflow</span>.<span class="pl-s1">models</span>.<span class="pl-s1">param</span> <span class="pl-k">import</span> <span class="pl-v">Param</span>

<span class="pl-s1">default_args</span> <span class="pl-c1">=</span> {
    <span class="pl-s">"owner"</span>: <span class="pl-s">"Airflow"</span>,
    <span class="pl-s">"depends_on_past"</span>: <span class="pl-c1">False</span>,
    <span class="pl-s">"start_date"</span>: <span class="pl-en">days_ago</span>(<span class="pl-c1">0</span>),
    <span class="pl-s">"email"</span>: [<span class="pl-s">"airflow@example.com"</span>],
    <span class="pl-s">"email_on_failure"</span>: <span class="pl-c1">False</span>,
    <span class="pl-s">"email_on_retry"</span>: <span class="pl-c1">False</span>,
    <span class="pl-s">"retries"</span>: <span class="pl-c1">0</span>,
    <span class="pl-s">"retry_delay"</span>: <span class="pl-en">timedelta</span>(<span class="pl-s1">minutes</span><span class="pl-c1">=</span><span class="pl-c1">5</span>),
}

<span class="pl-s1">dag</span> <span class="pl-c1">=</span> <span class="pl-v">DAG</span>(
    <span class="pl-s">"MapperK8sOperator-3"</span>,
    <span class="pl-s1">default_args</span><span class="pl-c1">=</span><span class="pl-s1">default_args</span>,
    <span class="pl-s1">schedule_interval</span><span class="pl-c1">=</span><span class="pl-en">timedelta</span>(<span class="pl-s1">minutes</span><span class="pl-c1">=</span><span class="pl-c1">30</span>),
    <span class="pl-s1">max_active_runs</span><span class="pl-c1">=</span><span class="pl-c1">1</span>,
    <span class="pl-s1">user_defined_filters</span><span class="pl-c1">=</span>{<span class="pl-s">"to_list"</span>: <span class="pl-k">lambda</span> <span class="pl-s1">x</span>: <span class="pl-en">list</span>(<span class="pl-s1">x</span>)},
    <span class="pl-s1">concurrency</span><span class="pl-c1">=</span><span class="pl-c1">10</span>,
    <span class="pl-s1">params</span><span class="pl-c1">=</span>{
        <span class="pl-s">"param1"</span>: <span class="pl-v">Param</span>(<span class="pl-c1">10</span>, <span class="pl-s1">type</span><span class="pl-c1">=</span><span class="pl-s">"integer"</span>, <span class="pl-s1">minimum</span><span class="pl-c1">=</span><span class="pl-c1">0</span>, <span class="pl-s1">maximum</span><span class="pl-c1">=</span><span class="pl-c1">20</span>),
        <span class="pl-s">"param2"</span>: <span class="pl-s">"value2"</span>,
    },
)

<span class="pl-c"># Generate 2 tasks</span>
<span class="pl-s1">tasks</span> <span class="pl-c1">=</span> [<span class="pl-s">"task{}"</span>.<span class="pl-en">format</span>(<span class="pl-s1">i</span>) <span class="pl-k">for</span> <span class="pl-s1">i</span> <span class="pl-c1">in</span> <span class="pl-en">range</span>(<span class="pl-c1">1</span>, <span class="pl-c1">3</span>)]
<span class="pl-s1">example_dag_complete_node</span> <span class="pl-c1">=</span> <span class="pl-v">BashOperator</span>(
    <span class="pl-s1">task_id</span><span class="pl-c1">=</span><span class="pl-s">"example_dag_complete"</span>,
    <span class="pl-s1">dag</span><span class="pl-c1">=</span><span class="pl-s1">dag</span>,
    <span class="pl-s1">bash_command</span><span class="pl-c1">=</span><span class="pl-s">"echo {{task_instance.xcom_pull(task_ids='task1',key='pod_name') | to_list }}"</span>,
)

<span class="pl-s1">org_dags</span> <span class="pl-c1">=</span> []


<span class="pl-k">def</span> <span class="pl-en">make_cmds</span>():
    <span class="pl-k">return</span> [
        [
            <span class="pl-s">"mkdir -p /airflow/xcom/;echo '[1,2,3,4,6]' &gt; /airflow/xcom/return.json;"</span>
        ],
        [
            <span class="pl-s">"mkdir -p /airflow/xcom/;echo '[1,2,3,4,109]' &gt; /airflow/xcom/return.json;"</span>
        ],
    ]


<span class="pl-k">for</span> <span class="pl-s1">task</span> <span class="pl-c1">in</span> <span class="pl-s1">tasks</span>:

    <span class="pl-s1">bash_command</span> <span class="pl-c1">=</span> <span class="pl-s">"echo HELLO"</span>

    <span class="pl-s1">org_node</span> <span class="pl-c1">=</span> <span class="pl-v">KubernetesPodOperator</span>.<span class="pl-en">partial</span>(
        <span class="pl-s1">namespace</span><span class="pl-c1">=</span><span class="pl-s">"my-namespace"</span>,
        <span class="pl-s1">image</span><span class="pl-c1">=</span><span class="pl-s">"python"</span>,
        <span class="pl-s1">labels</span><span class="pl-c1">=</span>{<span class="pl-s">"foo"</span>: <span class="pl-s">"bar"</span>},
        <span class="pl-s1">image_pull_policy</span><span class="pl-c1">=</span><span class="pl-s">"Always"</span>,
        <span class="pl-s1">name</span><span class="pl-c1">=</span><span class="pl-s1">task</span>,
        <span class="pl-s1">task_id</span><span class="pl-c1">=</span><span class="pl-s1">task</span>,
        <span class="pl-s1">is_delete_operator_pod</span><span class="pl-c1">=</span><span class="pl-c1">True</span>,
        <span class="pl-s1">resources</span> <span class="pl-c1">=</span> <span class="pl-s1">client</span>.<span class="pl-v">V1ResourceRequirements</span>(<span class="pl-s1">requests</span><span class="pl-c1">=</span>{
            <span class="pl-s">"cpu"</span>: <span class="pl-c1">1</span>,
            <span class="pl-s">"memory"</span>: <span class="pl-s">"256M"</span>,
        }),
        <span class="pl-s1">do_xcom_push</span><span class="pl-c1">=</span><span class="pl-c1">True</span>,
        <span class="pl-s1">get_logs</span><span class="pl-c1">=</span><span class="pl-c1">True</span>,
        <span class="pl-s1">dag</span><span class="pl-c1">=</span><span class="pl-s1">dag</span>,
    ).<span class="pl-en">expand</span>(<span class="pl-s1">cmds</span><span class="pl-c1">=</span><span class="pl-en">make_cmds</span>())

    <span class="pl-s1">org_node</span>.<span class="pl-en">set_downstream</span>(<span class="pl-s1">example_dag_complete_node</span>)
    <span class="pl-s1">org_node</span> <span class="pl-c1">&gt;&gt;</span> <span class="pl-s1">example_dag_complete_node</span></pre></div>
<h3 dir="auto">Operating System</h3>
<p dir="auto">Debian GNU/Linux</p>
<h3 dir="auto">Versions of Apache Airflow Providers</h3>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="apache-airflow-providers-amazon==3.4.0
apache-airflow-providers-celery==2.1.4
apache-airflow-providers-cncf-kubernetes==4.0.2
apache-airflow-providers-docker==2.7.0
apache-airflow-providers-elasticsearch==3.0.3
apache-airflow-providers-ftp==2.1.2
apache-airflow-providers-google==7.0.0
apache-airflow-providers-grpc==2.0.4
apache-airflow-providers-hashicorp==2.2.0
apache-airflow-providers-http==2.1.2
apache-airflow-providers-imap==2.2.3
apache-airflow-providers-microsoft-azure==3.9.0
apache-airflow-providers-mysql==2.2.3
apache-airflow-providers-odbc==2.0.4
apache-airflow-providers-postgres==4.1.0
apache-airflow-providers-redis==2.0.4
apache-airflow-providers-sendgrid==2.0.4
apache-airflow-providers-sftp==2.6.0
apache-airflow-providers-slack==4.2.3
apache-airflow-providers-sqlite==2.1.3
apache-airflow-providers-ssh==2.4.4"><pre class="notranslate"><code class="notranslate">apache-airflow-providers-amazon==3.4.0
apache-airflow-providers-celery==2.1.4
apache-airflow-providers-cncf-kubernetes==4.0.2
apache-airflow-providers-docker==2.7.0
apache-airflow-providers-elasticsearch==3.0.3
apache-airflow-providers-ftp==2.1.2
apache-airflow-providers-google==7.0.0
apache-airflow-providers-grpc==2.0.4
apache-airflow-providers-hashicorp==2.2.0
apache-airflow-providers-http==2.1.2
apache-airflow-providers-imap==2.2.3
apache-airflow-providers-microsoft-azure==3.9.0
apache-airflow-providers-mysql==2.2.3
apache-airflow-providers-odbc==2.0.4
apache-airflow-providers-postgres==4.1.0
apache-airflow-providers-redis==2.0.4
apache-airflow-providers-sendgrid==2.0.4
apache-airflow-providers-sftp==2.6.0
apache-airflow-providers-slack==4.2.3
apache-airflow-providers-sqlite==2.1.3
apache-airflow-providers-ssh==2.4.4
</code></pre></div>
<h3 dir="auto">Deployment</h3>
<p dir="auto">Official Apache Airflow Helm Chart</p>
<h3 dir="auto">Deployment details</h3>
<p dir="auto">The same example given here works without the <code class="notranslate">resources</code> keyword argument.</p>
<h3 dir="auto">Anything else</h3>
<p dir="auto"><em>No response</em></p>
<h3 dir="auto">Are you willing to submit PR?</h3>
<ul class="contains-task-list">
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox"> Yes I am willing to submit a PR!</li>
</ul>
<h3 dir="auto">Code of Conduct</h3>
<ul class="contains-task-list">
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox" checked=""> I agree to follow this project's <a href="https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md">Code of Conduct</a></li>
</ul>
<h3 dir="auto">Apache Airflow version</h3>
<p dir="auto">2.3.0 (latest released)</p>
<h3 dir="auto">What happened</h3>
<p dir="auto">When performing dynamic task mapping and providing Kubernetes limits to the <code class="notranslate">resources</code> argument, the DAG raises an import error:</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="Broken DAG: [/opt/airflow/dags/bulk_image_processing.py] Traceback (most recent call last):
  File &quot;/home/airflow/.local/lib/python3.7/site-packages/airflow/models/baseoperator.py&quot;, line 287, in partial
    partial_kwargs[&quot;resources&quot;] = coerce_resources(partial_kwargs[&quot;resources&quot;])
  File &quot;/home/airflow/.local/lib/python3.7/site-packages/airflow/models/baseoperator.py&quot;, line 133, in coerce_resources
    return Resources(**resources)
TypeError: __init__() got an unexpected keyword argument 'limit_cpu'"><pre class="notranslate"><code class="notranslate">Broken DAG: [/opt/airflow/dags/bulk_image_processing.py] Traceback (most recent call last):
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/models/baseoperator.py", line 287, in partial
    partial_kwargs["resources"] = coerce_resources(partial_kwargs["resources"])
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/models/baseoperator.py", line 133, in coerce_resources
    return Resources(**resources)
TypeError: __init__() got an unexpected keyword argument 'limit_cpu'
</code></pre></div>
<p dir="auto">The offending code is:</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="KubernetesPodOperator.partial(
            get_logs: True,
            in_cluster: True,
            is_delete_operator_pod: True,
            namespace: settings.namespace,
            resources={'limit_cpu': settings.IMAGE_PROCESSING_OPERATOR_CPU, 'limit_memory': settings.IMAGE_PROCESSING_OPERATOR_MEMORY},
            service_account_name: settings.SERVICE_ACCOUNT_NAME,
            startup_timeout_seconds: 600,
            **kwargs,
    )"><pre class="notranslate"><code class="notranslate">KubernetesPodOperator.partial(
            get_logs: True,
            in_cluster: True,
            is_delete_operator_pod: True,
            namespace: settings.namespace,
            resources={'limit_cpu': settings.IMAGE_PROCESSING_OPERATOR_CPU, 'limit_memory': settings.IMAGE_PROCESSING_OPERATOR_MEMORY},
            service_account_name: settings.SERVICE_ACCOUNT_NAME,
            startup_timeout_seconds: 600,
            **kwargs,
    )
</code></pre></div>
<p dir="auto">But you can see this in any DAG utilizing a <code class="notranslate">KubernetesPodOperator.partial</code> where the <code class="notranslate">partial</code> contains the <code class="notranslate">resources</code> argument.</p>
<h3 dir="auto">What you think should happen instead</h3>
<p dir="auto">The <code class="notranslate">resources</code> argument should be taken at face value and applied to the <code class="notranslate">OperatorPartial</code> and subsequently the <code class="notranslate">MappedOperator</code>.</p>
<h3 dir="auto">How to reproduce</h3>
<p dir="auto">Try to import this DAG using Airflow 2.3.0:</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="from datetime import datetime
from airflow import XComArg
from airflow.models import DAG
from airflow.providers.cncf.kubernetes.operators.kubernetes_pod import KubernetesPodOperator


def make_operator(
    **kwargs
):
    return KubernetesPodOperator(
        **{
            'get_logs': True,
            'in_cluster': True,
            'is_delete_operator_pod': True,
            'namespace': 'default',
            'startup_timeout_seconds': 600,
            **kwargs,
        }
    )


def make_partial_operator(
    **kwargs
):
    return KubernetesPodOperator.partial(
        **{
            'get_logs': True,
            'in_cluster': True,
            'is_delete_operator_pod': True,
            'namespace': 'default',
            'startup_timeout_seconds': 600,
            **kwargs,
        }
    )


with DAG(dag_id='bulk_image_processing',
         schedule_interval=None,
         start_date=datetime(2020, 1, 1),
         max_active_tasks=20) as dag:

    op1 = make_operator(
        arguments=['--bucket-name', f'{{{{ dag_run.conf.get(&quot;bucket&quot;, &quot;some-fake-default&quot;) }}}}'],
        cmds=['python3', 'some_entrypoint'],
        image='some-image',
        name='airflow-private-image-pod-1',
        task_id='some-task',
        do_xcom_push=True
    )

    op2 = make_partial_operator(
        image='another-image',
        name=f'airflow-private-image-pod-2',
        resources={'limit_cpu': '2000m', 'limit_memory': '16Gi'},
        task_id='another-task',
        cmds=[
            'some',
            'set',
            'of',
            'cmds'
        ]
    ).expand(arguments=XComArg(op1))
"><pre class="notranslate"><code class="notranslate">from datetime import datetime
from airflow import XComArg
from airflow.models import DAG
from airflow.providers.cncf.kubernetes.operators.kubernetes_pod import KubernetesPodOperator


def make_operator(
    **kwargs
):
    return KubernetesPodOperator(
        **{
            'get_logs': True,
            'in_cluster': True,
            'is_delete_operator_pod': True,
            'namespace': 'default',
            'startup_timeout_seconds': 600,
            **kwargs,
        }
    )


def make_partial_operator(
    **kwargs
):
    return KubernetesPodOperator.partial(
        **{
            'get_logs': True,
            'in_cluster': True,
            'is_delete_operator_pod': True,
            'namespace': 'default',
            'startup_timeout_seconds': 600,
            **kwargs,
        }
    )


with DAG(dag_id='bulk_image_processing',
         schedule_interval=None,
         start_date=datetime(2020, 1, 1),
         max_active_tasks=20) as dag:

    op1 = make_operator(
        arguments=['--bucket-name', f'{{{{ dag_run.conf.get("bucket", "some-fake-default") }}}}'],
        cmds=['python3', 'some_entrypoint'],
        image='some-image',
        name='airflow-private-image-pod-1',
        task_id='some-task',
        do_xcom_push=True
    )

    op2 = make_partial_operator(
        image='another-image',
        name=f'airflow-private-image-pod-2',
        resources={'limit_cpu': '2000m', 'limit_memory': '16Gi'},
        task_id='another-task',
        cmds=[
            'some',
            'set',
            'of',
            'cmds'
        ]
    ).expand(arguments=XComArg(op1))

</code></pre></div>
<h3 dir="auto">Operating System</h3>
<p dir="auto">MacOS 11.6.5</p>
<h3 dir="auto">Versions of Apache Airflow Providers</h3>
<p dir="auto">Relevant:</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="apache-airflow-providers-cncf-kubernetes==4.0.1"><pre class="notranslate"><code class="notranslate">apache-airflow-providers-cncf-kubernetes==4.0.1
</code></pre></div>
<p dir="auto">Full:</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="apache-airflow-providers-amazon==3.3.0
apache-airflow-providers-celery==2.1.4
apache-airflow-providers-cncf-kubernetes==4.0.1
apache-airflow-providers-docker==2.6.0
apache-airflow-providers-elasticsearch==3.0.3
apache-airflow-providers-ftp==2.1.2
apache-airflow-providers-google==6.8.0
apache-airflow-providers-grpc==2.0.4
apache-airflow-providers-hashicorp==2.2.0
apache-airflow-providers-http==2.1.2
apache-airflow-providers-imap==2.2.3
apache-airflow-providers-microsoft-azure==3.8.0
apache-airflow-providers-mysql==2.2.3
apache-airflow-providers-odbc==2.0.4
apache-airflow-providers-postgres==4.1.0
apache-airflow-providers-redis==2.0.4
apache-airflow-providers-sendgrid==2.0.4
apache-airflow-providers-sftp==2.6.0
apache-airflow-providers-slack==4.2.3
apache-airflow-providers-sqlite==2.1.3
apache-airflow-providers-ssh==2.4.3"><pre class="notranslate"><code class="notranslate">apache-airflow-providers-amazon==3.3.0
apache-airflow-providers-celery==2.1.4
apache-airflow-providers-cncf-kubernetes==4.0.1
apache-airflow-providers-docker==2.6.0
apache-airflow-providers-elasticsearch==3.0.3
apache-airflow-providers-ftp==2.1.2
apache-airflow-providers-google==6.8.0
apache-airflow-providers-grpc==2.0.4
apache-airflow-providers-hashicorp==2.2.0
apache-airflow-providers-http==2.1.2
apache-airflow-providers-imap==2.2.3
apache-airflow-providers-microsoft-azure==3.8.0
apache-airflow-providers-mysql==2.2.3
apache-airflow-providers-odbc==2.0.4
apache-airflow-providers-postgres==4.1.0
apache-airflow-providers-redis==2.0.4
apache-airflow-providers-sendgrid==2.0.4
apache-airflow-providers-sftp==2.6.0
apache-airflow-providers-slack==4.2.3
apache-airflow-providers-sqlite==2.1.3
apache-airflow-providers-ssh==2.4.3
</code></pre></div>
<h3 dir="auto">Deployment</h3>
<p dir="auto">Official Apache Airflow Helm Chart</p>
<h3 dir="auto">Deployment details</h3>
<p dir="auto">Docker (Docker Desktop)</p>
<ul dir="auto">
<li>Server Version: 20.10.13</li>
<li>API Version: 1.41</li>
<li>Builder: 2</li>
</ul>
<p dir="auto">Kubernetes (Docker Desktop)</p>
<ul dir="auto">
<li>Env: docker-desktop</li>
<li>Context: docker-desktop</li>
<li>Cluster Name: docker-desktop</li>
<li>Namespace: default</li>
<li>Container Runtime: docker</li>
<li>Version: v1.22.5</li>
</ul>
<p dir="auto">Helm:</p>
<ul dir="auto">
<li>version.BuildInfo{Version:"v3.6.3", GitCommit:"d506314abfb5d21419df8c7e7e68012379db2354", GitTreeState:"dirty", GoVersion:"go1.16.5"}</li>
</ul>
<h3 dir="auto">Anything else</h3>
<p dir="auto">You can get around this by creating the <code class="notranslate">partial</code> first without calling <code class="notranslate">expand</code> on it, setting the resources via the <code class="notranslate">kwargs</code> parameter, then calling <code class="notranslate">expand</code>. Example:</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="from datetime import datetime
from airflow import XComArg
from airflow.models import DAG
from airflow.providers.cncf.kubernetes.operators.kubernetes_pod import KubernetesPodOperator


def make_operator(
    **kwargs
):
    return KubernetesPodOperator(
        **{
            'get_logs': True,
            'in_cluster': True,
            'is_delete_operator_pod': True,
            'namespace': 'default',
            'startup_timeout_seconds': 600,
            **kwargs,
        }
    )


def make_partial_operator(
    **kwargs
):
    return KubernetesPodOperator.partial(
        **{
            'get_logs': True,
            'in_cluster': True,
            'is_delete_operator_pod': True,
            'namespace': 'default',
            'startup_timeout_seconds': 600,
            **kwargs,
        }
    )


with DAG(dag_id='bulk_image_processing',
         schedule_interval=None,
         start_date=datetime(2020, 1, 1),
         max_active_tasks=20) as dag:

    op1 = make_operator(
        arguments=['--bucket-name', f'{{{{ dag_run.conf.get(&quot;bucket&quot;, &quot;some-fake-default&quot;) }}}}'],
        cmds=['python3', 'some_entrypoint'],
        image='some-image',
        name='airflow-private-image-pod-1',
        task_id='some-task',
        do_xcom_push=True
    )

    op2 = make_partial_operator(
        image='another-image',
        name=f'airflow-private-image-pod-2',
        task_id='another-task',
        cmds=[
            'some',
            'set',
            'of',
            'cmds'
        ]
    )
    
    op2.kwargs['resources'] = {'limit_cpu': '2000m', 'limit_memory': '16Gi'} 

    op2.expand(arguments=XComArg(op1))"><pre class="notranslate"><code class="notranslate">from datetime import datetime
from airflow import XComArg
from airflow.models import DAG
from airflow.providers.cncf.kubernetes.operators.kubernetes_pod import KubernetesPodOperator


def make_operator(
    **kwargs
):
    return KubernetesPodOperator(
        **{
            'get_logs': True,
            'in_cluster': True,
            'is_delete_operator_pod': True,
            'namespace': 'default',
            'startup_timeout_seconds': 600,
            **kwargs,
        }
    )


def make_partial_operator(
    **kwargs
):
    return KubernetesPodOperator.partial(
        **{
            'get_logs': True,
            'in_cluster': True,
            'is_delete_operator_pod': True,
            'namespace': 'default',
            'startup_timeout_seconds': 600,
            **kwargs,
        }
    )


with DAG(dag_id='bulk_image_processing',
         schedule_interval=None,
         start_date=datetime(2020, 1, 1),
         max_active_tasks=20) as dag:

    op1 = make_operator(
        arguments=['--bucket-name', f'{{{{ dag_run.conf.get("bucket", "some-fake-default") }}}}'],
        cmds=['python3', 'some_entrypoint'],
        image='some-image',
        name='airflow-private-image-pod-1',
        task_id='some-task',
        do_xcom_push=True
    )

    op2 = make_partial_operator(
        image='another-image',
        name=f'airflow-private-image-pod-2',
        task_id='another-task',
        cmds=[
            'some',
            'set',
            'of',
            'cmds'
        ]
    )
    
    op2.kwargs['resources'] = {'limit_cpu': '2000m', 'limit_memory': '16Gi'} 

    op2.expand(arguments=XComArg(op1))
</code></pre></div>
<h3 dir="auto">Are you willing to submit PR?</h3>
<ul class="contains-task-list">
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox"> Yes I am willing to submit a PR!</li>
</ul>
<h3 dir="auto">Code of Conduct</h3>
<ul class="contains-task-list">
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox" checked=""> I agree to follow this project's <a href="https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md">Code of Conduct</a></li>
</ul>