当Airflow任务通过@task.kubernetes()装饰器在Kubernetes Pod中运行时，它在一个独立且隔离的环境中执行。默认情况下，如果指定了如image="python"这样的基础镜像，该Pod将只包含基础Python环境，而不包含任何在Airflow调度器或Web服务器环境中安装的第三方库，也无法访问DAG文件同目录下的自定义模块。这会导致常见的NameError或ModuleNotFoundError错误。解决此问题的关键在于两个核心步骤：创建包含所有依赖的自定义Docker镜像，并将模块导入语句放置在Kubernetes任务函数内部。
1. 构建包含依赖的自定义Docker镜像
为了让Kubernetes Pod中的任务能够访问所需的第三方库和自定义代码，我们需要创建一个定制的Docker镜像。这个镜像将作为@task.kubernetes()任务的基础运行环境。
1.1 准备Dockerfile
在项目的根目录或一个专门的目录中，创建Dockerfile文件，并添加以下内容：
# 基于一个官方的Python镜像
FROM python:3.10.0
# 创建一个目录用于存放自定义模块
RUN mkdir /mymodule
# 将本地的mymodule目录复制到镜像的/mymodule中
# 假设你的自定义模块代码位于名为'mymodule'的文件夹中
COPY mymodule /mymodule
# 安装所有第三方依赖
# 确保在项目根目录有一个requirements.txt文件，列出所有Python依赖
COPY requirements.txt .
RUN pip install -r requirements.txt
# 将/mymodule添加到PYTHONPATH，使其内部的模块可以被导入
ENV PYTHONPATH "${PYTHONPATH}:/mymodule"Dockerfile说明：
- FROM python:3.10.0: 选择一个适合你项目的基础Python镜像。
- RUN mkdir /mymodule: 在镜像中创建一个目录，用于存放你的自定义模块。
- COPY mymodule /mymodule: 将你本地的mymodule文件夹（包含process_data等自定义函数）复制到镜像的/mymodule路径下。
- COPY requirements.txt . 和 RUN pip install -r requirements.txt: 将你的requirements.txt文件复制到镜像中，并安装所有列出的第三方Python包。确保requirements.txt包含了如decouple等所有必要的外部依赖。
- ENV PYTHONPATH "${PYTHONPATH}:/mymodule": 这一步至关重要。它将/mymodule路径添加到Python的搜索路径中，使得在任务函数内部可以通过from mymodule import ...来导入你的自定义模块。
1.2 构建与推送Docker镜像
在包含Dockerfile的目录下执行以下命令来构建你的Docker镜像：
docker build -t your_image_with_mymodule:latest .
- your_image_with_mymodule: 替换为你自定义的镜像名称。
- :latest: 建议使用有意义的版本标签代替latest，例如v1.0.0。
如果你的Airflow部署在云环境中（如GKE、AWS EKS），你需要将此镜像推送到一个可访问的Docker镜像仓库（如Google Container Registry (GCR), Amazon Elastic Container Registry (ECR), Docker Hub等）。
docker tag your_image_with_mymodule:latest your_registry/your_image_with_mymodule:latest docker push your_registry/your_image_with_mymodule:latest
如果Airflow运行在本地环境，且Docker守护进程可以访问到你本地构建的镜像，则无需推送。
2. 更新Airflow DAG以使用自定义镜像和内部导入
构建并推送完自定义镜像后，你需要修改Airflow DAG，使其指向新创建的镜像，并将所有相关的导入语句移动到@task.kubernetes()装饰的任务函数内部。
from airflow.decorators import dag, task
from datetime import datetime
@dag(
"model_trainer",
start_date=datetime(2023, 1, 1),
catchup=False,
schedule=None,
tags=["kubernetes", "dependencies"],
)
def pipeline():
@task.kubernetes(
image="your_image_with_mymodule:latest", # 使用你构建的自定义镜像
# 其他Kubernetes相关的参数，例如资源限制、命名空间等
# namespace="airflow",
# do_xcom_push=True,
# get_logs=True,
)
def fetch_data():
# 将所有自定义模块和第三方库的导入语句移动到函数内部
from mymodule import process_data
# from decouple import AutoConfig # 如果AutoConfig未在函数内部使用，可以删除此行
# 执行实际的数据处理逻辑
result = process_data()
print(f"Data processed: {result}")
return result
# 实例化任务
fetch_data_task = fetch_data()
# 实例化DAG
pipeline()更新说明：
- @task.kubernetes(image="your_image_with_mymodule:latest"): 将image参数的值更改为你刚刚构建并可能已推送的自定义Docker镜像的完整路径（包括仓库地址和标签）。
- 导入语句移动: 将from mymodule import process_data和任何其他第三方库（如from decouple import AutoConfig）的导入语句从DAG文件的顶部移动到fetch_data()函数内部。
为什么导入语句需要移动到函数内部？
- 环境隔离: task.kubernetes任务在独立的Pod中运行，其Python环境与Airflow调度器/Web服务器的环境是完全隔离的。DAG文件顶部的导入是在调度器解析DAG时执行的，它使用的是调度器的Python环境。而任务函数内部的导入，则是在Pod启动后，任务代码执行时，使用的是Pod内部的Python环境。
- 避免冲突与冗余: 这样做可以确保每个Kubernetes任务只加载其真正需要的依赖，避免不必要的库加载和潜在的版本冲突。
3. 注意事项与最佳实践
- 依赖管理: 始终使用requirements.txt来管理第三方Python依赖。这使得Docker镜像的构建过程更加自动化和可重复。
- 镜像版本控制: 为你的Docker镜像使用有意义的版本标签（例如v1.0.0, 20230101-featureX），而不是仅仅使用latest。这有助于回滚和管理不同版本的任务环境。
- 本地测试: 在将DAG部署到生产环境之前，先在本地测试你的Docker镜像和DAG。你可以使用docker run命令来测试镜像是否正确包含了所有依赖，并使用airflow dags test your_dag_id来测试DAG的语法和任务逻辑。
- 资源限制: 在@task.kubernetes()中配置适当的资源限制（CPU、内存），以防止Pod消耗过多资源或因资源不足而失败。
- 日志与XCom: 确保get_logs=True以便于调试，并合理利用XCom进行任务间的数据传递。
通过遵循上述步骤，你可以有效地在Airflow中使用@task.kubernetes()装饰器来运行包含第三方和自定义依赖的任务，确保它们在隔离的Kubernetes环境中稳定可靠地执行。
