Airflow环境变量为空求助：如何传递Spotify播放列表ID

问题根源 #

1. 你写入.env的变量名是AIRFLOW_VAR_PLAYLIST_ID，但Docker Compose里引用的是${PLAYLIST_ID}，两者不匹配，导致容器无法获取到正确值
2. BashOperator中没有正确传递Python侧获取的变量，模板变量使用方式有误

分步解决方案 #

1. 修正.env变量名匹配

修改你的Python脚本，让写入.env的变量名与Docker Compose中引用的一致：

def get_input(prompt):
    return input(prompt)

playlist_id = get_input("Bitte geben Sie die PLAYLIST_ID ein: ")

with open('.env', 'w') as f:
    # 改为PLAYLIST_ID，和Docker Compose引用的变量名对齐
    f.write(f"PLAYLIST_ID={playlist_id}\n")

2. 确保Docker Compose正确加载.env

Docker Compose默认读取当前目录的.env文件，确认docker-compose.yml所在目录下有修改后的.env，Airflow服务配置保持：

services:
  airflow:
    build:
      context: .
      dockerfile: Dockerfile.airflow
    container_name: airflow
    environment:
      # 现在.env里有PLAYLIST_ID，可正确注入容器的AIRFLOW_VAR_PLAYLIST_ID变量
      - AIRFLOW_VAR_PLAYLIST_ID=${PLAYLIST_ID}
    volumes:
      ...

注意：重启容器前执行docker-compose down && docker-compose up -d，避免缓存旧环境变量。

3. 在DAG中正确传递变量到BashOperator

有两种可靠实现方式：

方式一：直接传递环境变量

利用BashOperator的env参数，把Python侧获取的变量传入bash执行环境：

import os
from airflow.operators.bash import BashOperator

# 从容器环境变量中获取值
playlist_id = os.getenv("AIRFLOW_VAR_PLAYLIST_ID")

get_playlist_tracks = BashOperator(
    task_id='get_playlist_tracks',
    bash_command="""
    echo "AIRFLOW_VAR_PLAYLIST_ID = $AIRFLOW_VAR_PLAYLIST_ID"
    echo "PLAYLIST_ID = $PLAYLIST_ID"
    curl -X "GET" "https://api.spotify.com/v1/playlists/$PLAYLIST_ID/tracks" \
        -H "Accept: application/json" \
        -H "Content-Type: application/json" \
        -H "Authorization: Bearer $ACCESS_TOKEN" \
        > /user/hadoop/spotify/track_data/raw/playlist_tracks.json
    echo "Playlist tracks fetched and saved to HDFS"
    """,
    # 传递变量到bash环境
    env={
        "PLAYLIST_ID": playlist_id,
        "ACCESS_TOKEN": os.getenv("ACCESS_TOKEN")  # 若ACCESS_TOKEN也是环境变量
    },
    dag=dag 
)

方式二：使用Airflow变量系统（推荐）

将PLAYLIST_ID存入Airflow Variables，通过模板引用，更符合Airflow最佳实践：

1. 先存入变量（可通过脚本、UI或CLI）：

from airflow.models import Variable

# 假设已获取到playlist_id
Variable.set("PLAYLIST_ID", playlist_id)

2. 在DAG的BashOperator中使用模板：

from airflow.operators.bash import BashOperator

get_playlist_tracks = BashOperator(
    task_id='get_playlist_tracks',
    bash_command="""
    echo "PLAYLIST_ID = {{ var.value.PLAYLIST_ID }}"
    curl -X "GET" "https://api.spotify.com/v1/playlists/{{ var.value.PLAYLIST_ID }}/tracks" \
        -H "Accept: application/json" \
        -H "Content-Type: application/json" \
        -H "Authorization: Bearer {{ var.value.ACCESS_TOKEN }}" \
        > /user/hadoop/spotify/track_data/raw/playlist_tracks.json
    echo "Playlist tracks fetched and saved to HDFS"
    """,
    dag=dag 
)

额外排查点 #

• 进入Airflow容器执行echo $AIRFLOW_VAR_PLAYLIST_ID，确认变量已正确注入
• 检查.env文件权限，确保Docker Compose能读取
• 若使用Airflow 2.x，确认无限制变量加载的配置项

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