主页 / user-14275533

lalaland's questions

Martin Hope
lalaland
Asked: 2025-04-11 09:07:32 +0800 CST
  • 5

我一直尝试使用 helm chart（作为子图）在我的家庭实验室中托管的 kubernetes 集群上部署 airflow。

我在将本地测试的 dag（在 docker-compose 上）迁移到 Kubernetes 集群时遇到了一些问题。在 Kubernetes 上使用 Airflow 时，可以使用 CeleryExecutor 吗？如果可以，我应该如何更改 dag 中的 DockerOperators？在寻找答案的过程中，我偶然发现了以下 stackoverflow 帖子，它建议在 Kubernetes 上运行时使用 DockerOperator 不是一个好选择，对吗？

  • 这是否意味着我不应该使用 CeleryExecutor ？

  • 将 dags 从 docker-compose 迁移到 Kubernetes 时，最佳实践是什么？谢谢

docker
Martin Hope
lalaland
Asked: 2024-11-10 14:04:44 +0800 CST
  • 5

我正在创建一个应该执行以下操作的 dag：

  • 获取事件 ID
  • 对于每个事件 ID，获取事件详细信息 (DockerOperator)

下面的代码是我尝试做我想做的事情：

import re
from datetime import datetime

from airflow.decorators import dag, task
from airflow.hooks.base import BaseHook
from airflow.models import Variable
from airflow.models.baseoperator import chain
from airflow.operators.python import PythonOperator
from airflow.providers.docker.operators.docker import DockerOperator
from airflow.providers.slack.notifications.slack import SlackNotifier
from airflow.sensors.base import PokeReturnValue


leagues = ["league1", "league2", "league3"]




@dag(
    start_date=datetime(2024, 11, 1),
    schedule="@daily",
)
    task_fetch_ids = PythonOperator(
        task_id="fetch_detail",
        ...)


    task_fetch_detail = DockerOperator(
        task_id="fetch_detail",
        image="image:v1",
        ).expand(
            command=[f"fetch-event --event-id  {event_id}" for event_id in "{{ ti.xcom_pull(task_ids='task_fetch_ids', key='return_value') }}"]
        )


    task_fetch_ids >> task_fetch_detail

上面的代码显然不起作用，因为我正在循环遍历一个字符串。正确的语法是什么？

python