Airflow中KubernetesPodOperator调用Rscript提示命令未找到

已成功构建包含R的Docker镜像，但在Composer管理的Airflow环境中，通过KubernetesPodOperator调用时仍出现Rscript: command not found错误。

错误日志 #

[2023-11-02, 01:15:40 UTC] {subprocess.py:63} INFO - Tmp dir root location: 
 /tmp
[2023-11-02, 01:15:40 UTC] {subprocess.py:75} INFO - Running command: ['/usr/bin/bash', '-c', 'Rscript GCS_PATH_TO_R_FILE/prueba-bash.r']
[2023-11-02, 01:15:41 UTC] {subprocess.py:86} INFO - Output:
[2023-11-02, 01:15:41 UTC] {subprocess.py:93} INFO - /usr/bin/bash: Rscript: command not found
[2023-11-02, 01:15:41 UTC] {subprocess.py:97} INFO - Command exited with return code 127

Dockerfile内容 #

FROM ubuntu:latest

ENV DEBIAN_FRONTEND=noninteractive

RUN apt-get update && apt-get install -y --no-install-recommends build-essential \
    r-base r-base-core r-cran-randomforest \
    libinput-dev libgbm-dev liblzma-dev libbz2-dev libicu-dev libblas-dev liblapack-dev \
    python3.6 python3-pip python3-setuptools python3-dev&& \
   rm -r /var/lib/apt/lists/*

WORKDIR /app

COPY requirements.txt /app/requirements.txt

RUN pip3 install --upgrade pip wheel setuptools>51

RUN python3 -m pip install rpy2

RUN Rscript -e "install.packages('data.table')"

COPY . /app

原DAG代码 #

from airflow import models
from airflow.contrib.operators.kubernetes_pod_operator import KubernetesPodOperator
from airflow import DAG
from airflow.operators.bash import BashOperator
from datetime import datetime, timedelta

import os 

# Get current directory
cwd = os.getcwd()

default_dag_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime(2023, 10, 31),
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(seconds=5),
}

dag = DAG(
    'prueba-bash-gcs',
    default_args=default_dag_args,
    schedule_interval=None,
)


kubernetes_min_pod = KubernetesPodOperator(
    task_id="test-bash-r",
    name="test-bash-r",
    cmds=["echo"],
    namespace="default",
    image="gcr.io/ldp-data-wyscout/ambienterypython:v1",
)

test_bash = BashOperator(
    task_id='test-bash-r',
    dag = dag,
    bash_command = f'Rscript GCS_PATH_TO_R_FILE/prueba-bash.r'
)

问题根源与解决方案 #

核心问题 #

你误解了Airflow task_id的作用：BashOperator是在Airflow Worker的容器中运行的，而非KubernetesPodOperator指定的自定义镜像。所以即使定义了KubernetesPodOperator，BashOperator依然会在没有安装R的Worker容器里执行命令，自然找不到Rscript。

解决步骤 #

1. 移除多余的BashOperator，直接在KubernetesPodOperator中配置执行R脚本的命令
2. 确保GCS文件可被容器访问：通过Volume将GCS路径挂载到容器内部（比如使用GCS FUSE），或者提前将R脚本复制到镜像中
3. 使用绝对路径执行Rscript：避免环境变量问题，可提前在镜像中用which Rscript确认路径（通常为/usr/bin/Rscript）

修改后的DAG代码示例 #

from airflow import DAG
from airflow.contrib.operators.kubernetes_pod_operator import KubernetesPodOperator
from datetime import datetime, timedelta

default_dag_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime(2023, 10, 31),
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(seconds=5),
}

dag = DAG(
    'prueba-bash-gcs',
    default_args=default_dag_args,
    schedule_interval=None,
)

# 直接在KubernetesPodOperator中执行R脚本
run_r_script = KubernetesPodOperator(
    task_id="run-r-script",
    name="run-r-script",
    # 配置执行命令，替换为容器内可访问的R脚本路径
    cmds=["/usr/bin/Rscript", "/path/in/container/prueba-bash.r"],
    namespace="default",
    image="gcr.io/ldp-data-wyscout/ambienterypython:v1",
    # 如需挂载GCS，添加volume和volume_mounts配置
    # volumes=[...],
    # volume_mounts=[...]
)

内容的提问来源于stack exchange，提问作者moltke_colombia