如何从使用 KubernetesPodOperator 触发它的 Airflow 主机将卷挂载到运行 docker 容器的 Kubernetes Pod

Question

我在使用 KubernetesPodOperator 的气流中有一个 DAG，我正在尝试将在 pod 中运行的容器生成的一些文件返回到气流主机。对于开发，我的主机是一个运行 airflow image 和 docker-desktop K8s 集群的 Docker 容器，对于生产，我使用带有 EKS 的 AWS EC2 盒。

volume_mount = VolumeMount('dbt-home',
                           mount_path=<CONTAINER_DIR>,
                           sub_path=None,
                           read_only=False)

volume_config= {
    'hostPath':
      {'path': <HOST_DIR>, 'type': 'DirectoryOrCreate'}
    }

volume = Volume(name="dbt-home", configs=volume_config)


dbt_run = KubernetesPodOperator(
                          namespace='default',
                          image=MY_IMAGE>,
                          cmds=["bash", "-cx"],
                          arguments=[command],
                          env_vars=MY_ENVIRONMENT,
                          volumes=[volume],
                          volume_mounts=[volume_mount],
                          name="test-run",
                          task_id="test-run-task",
                          config_file=config_file,
                          get_logs=True,
                          reattach_on_restart=True,
                          dag=dag
                          )

我尝试对卷使用 hostPath 类型，但我认为它指的是 pod 的主机。我查看了有关卷的 kubernetes 文档，其中我发现 EmptyDir 也没有解决。

Answer 1

根据您的评论，您是在询问如何在 pod 中运行的一个任务如何完成并将日志写入到 pod 中运行的另一个任务在启动时可以读取的位置。看起来你可以做一些事情。

1. 您可以让您的任务开始获取通过 kubectl get logs 完成的前一个 pod 的日志（即将 kubectl 放入您的任务映像并允许其服务帐户获取该命名空间中的 pod 日志）或使用 Kubernetes python API 获取日志。
2. 您可以将 pvc 挂载到您的初始任务中的某个位置并在那里写入日志，然后当它完成时，您可以将相同的 pvc 挂载到您的下一个任务中，它可以从该位置读取日志。如果一次只能挂载到一个 pod 中，则可以使用 ebs；如果一次挂载到多个 pod 中，则可以使用 nfs。可能 nfs 会很有意义，这样您就可以一次在 pod 中的多个任务之间共享您的日志。
3. 您可以将日志发送至Cloudwatch via fluentd。然后，您的任务可以查询 Cloudwatch 以获取上一个任务的日志。我认为将您的日志传送到 Cloudwatch 是一种很好的做法，因此您不妨这样做。

我不确定您是否正在寻找一种更自然的气流方式来执行此操作，但这些想法可以解决您的问题。