在GCP Astronomer中使用Bash Operator访问GCS Bucket文件报错求助

可以通过Bash Operator访问GCS Bucket，但普通Linux命令（如awk）无法直接识别gs://协议路径，这就是你报错的原因——awk把云存储路径当成了本地文件，自然找不到。

解决方案 #

要处理GCS中的文件，需要借助Google Cloud SDK提供的gsutil工具（Airflow官方的GCS镜像一般已预装，若没有则需自行安装Google Cloud SDK），有两种常用方式：

方式1：下载文件到本地临时目录（适合大文件）

先将GCS文件下载到本地临时路径，再用awk等工具处理，完成后清理临时文件：

bash_operator = BashOperator(
    task_id='mani_bash',
    bash_command="""
# 创建临时文件
TEMP_FILE=$(mktemp)
# 从GCS下载文件到本地
gsutil cp gs://<bucketname>/<location>/filename.txt "$TEMP_FILE"

# 一次性提取所需数据，避免重复扫描文件
COL_STATS=$(awk -F: '/^[^HDR][^TRL]/ {print}' "$TEMP_FILE" | awk -F "|" '{print NF-1}' | uniq)
UNIQ_COUNT=$(echo "$COL_STATS" | wc -l)

if [ "$UNIQ_COUNT" -eq 1 ]; then
    if [ "$COL_STATS" -eq 9 ]; then
        echo 'rite'
    fi
else
    echo 'not rite'
fi

# 清理临时文件
rm "$TEMP_FILE"
""",
)

方式2：直接流式读取（适合小文件）

用gsutil cat将GCS文件内容直接输出到管道，无需下载整个文件，更高效：

bash_operator = BashOperator(
    task_id='mani_bash',
    bash_command="""
# 流式读取GCS文件内容并处理
COL_STATS=$(gsutil cat gs://<bucketname>/<location>/filename.txt | awk -F: '/^[^HDR][^TRL]/ {print}' | awk -F "|" '{print NF-1}' | uniq)
UNIQ_COUNT=$(echo "$COL_STATS" | wc -l)

if [ "$UNIQ_COUNT" -eq 1 ]; then
    if [ "$COL_STATS" -eq 9 ]; then
        echo 'rite'
    fi
else
    echo 'not rite'
fi
""",
)

关键注意事项 #

• 权限配置：确保执行Bash Operator的Airflow Worker拥有GCS Bucket的访问权限（可通过服务账号密钥、工作负载身份或IAM角色配置）。
• 临时文件管理：若运行在Kubernetes环境，需注意临时目录的存储卷权限，避免无法创建或写入临时文件。
• 脚本优化：原脚本重复读取了两次GCS文件，优化后仅扫描一次，提升了执行效率。

内容的提问来源于stack exchange，提问作者Mani Shankar.S