Airflow中用BashOperator执行Docker命令报错求助

问题背景 #

因环境问题无法使用DockerOperator，改用BashOperator运行Docker容器，示例代码如下：

run_model_op = BashOperator(
                task_id=f'run_model_GPU{gpu}',
                env={"CUDA_VISIBLE_DEVICES":gpu},
                bash_command='docker images',
                append_env=True,
                dag=dag,
            )

DAG采用多GPU并行结构：

start - run_model_op(gpu0)- run_model_op_3(gpu0)...  
      - run_model_op_1(gpu1)- run_model_op_2(gpu1)...  
      ...

报错信息 #

运行时出现如下错误：

[2023-04-07, 22:29:19 KST] {subprocess.py:74} INFO - Running command: ['bash', '-c', 'docker images']
[2023-04-07, 22:29:19 KST] {taskinstance.py:1909} ERROR - Task failed with exception
Traceback (most recent call last):
  File "/home/airflow/anaconda3/envs/airflow/lib/python3.8/site-packages/airflow/operators/bash.py", line 185, in execute
    result = self.subprocess_hook.run_command(
  File "/home/airflow/anaconda3/envs/airflow/lib/python3.8/site-packages/airflow/hooks/subprocess.py", line 76, in run_command
    self.sub_process = Popen(
  File "/home/airflow/anaconda3/envs/airflow/lib/python3.8/subprocess.py", line 858, in __init__
    self._execute_child(args, executable, preexec_fn, close_fds,
  File "/home/airflow/anaconda3/envs/airflow/lib/python3.8/subprocess.py", line 1626, in _execute_child
    env_list.append(k + b'=' + os.fsencode(v))
  File "/home/airflow/anaconda3/envs/airflow/lib/python3.8/os.py", line 806, in fsencode
    filename = fspath(filename)  # Does type-checking of `filename`.
TypeError: expected str, bytes or os.PathLike object, not int

问题原因 #

报错核心是TypeError: expected str, bytes or os.PathLike object, not int，根源在于环境变量的值必须是字符串/字节类型，但代码中gpu是整数类型（比如0、1这类数值），传递给CUDA_VISIBLE_DEVICES时未做类型转换，导致os.fsencode处理整数时抛出异常。

解决方案 #

将gpu转换为字符串类型后再传入环境变量，修改后的代码如下：

run_model_op = BashOperator(
                task_id=f'run_model_GPU{gpu}',
                env={"CUDA_VISIBLE_DEVICES": str(gpu)},  # 转为字符串类型
                bash_command='docker images',
                append_env=True,
                dag=dag,
            )

内容的提问来源于stack exchange，提问作者이준혁