Dataproc中PySpark如何通过Python logging直接写入GCS Bucket日志？

Python标准库的logging.basicConfig不支持直接写入GCS路径，因为它底层依赖本地文件系统的IO操作，无法识别gs://这种云存储URI，会把它解析成本地路径/gs:/bucket_name/newfile.log，自然找不到对应的目录和文件，导致FileNotFoundError。

以下几种方式可以实现将PySpark任务日志写入GCS Bucket：

1. 自定义GCS Logging Handler（推荐） #

借助Google Cloud Storage Python客户端库，实现自定义的logging Handler，直接将日志内容写入GCS对象。

步骤： #

• 确保集群已安装google-cloud-storage库（如果未预装，可通过Dataproc初始化动作或任务中执行pip install google-cloud-storage）
• 编写自定义Handler并配置logging：

import logging
from google.cloud import storage

class GCSLoggingHandler(logging.Handler):
    def __init__(self, bucket_name, blob_name, initial_clear=True):
        super().__init__()
        self.client = storage.Client()
        self.bucket = self.client.bucket(bucket_name)
        self.blob = self.bucket.blob(blob_name)
        # 初始化时清空目标文件（模拟filemode='w'）
        if initial_clear:
            self.blob.upload_from_string('')
        # 用内存缓冲区减少API调用次数，优化性能
        self.buffer = []

    def emit(self, record):
        log_entry = self.format(record) + '\n'
        self.buffer.append(log_entry)
        # 当缓冲区达到100条日志时批量写入，可根据需求调整
        if len(self.buffer) >= 100:
            self._flush_buffer()

    def _flush_buffer(self):
        if not self.buffer:
            return
        # 获取现有内容（如果需要追加）
        existing_content = self.blob.download_as_text() if self.blob.exists() else ''
        new_content = existing_content + ''.join(self.buffer)
        self.blob.upload_from_string(new_content)
        self.buffer.clear()

    def close(self):
        # 关闭Handler时刷新剩余缓冲区
        self._flush_buffer()
        super().close()

# 配置日志
logger = logging.getLogger("pyspark-gcs-logger")
logger.setLevel(logging.INFO)
logger.propagate = False  # 避免重复输出到控制台/YARN日志

# 创建并添加GCS Handler
gcs_handler = GCSLoggingHandler("bucket_name", "newfile.log")
log_formatter = logging.Formatter("%(asctime)s %(message)s")
gcs_handler.setFormatter(log_formatter)
logger.addHandler(gcs_handler)

# 测试日志输出
logger.info("PySpark任务日志已写入GCS")

注意事项： #

• 集群的服务账号需拥有GCS的storage.objects.create和storage.objects.update权限，确保能正常读写GCS对象
• 缓冲区大小可根据日志量调整，减少频繁的GCS API调用，提升性能
• 任务结束前需调用logger.handlers[0].close()手动刷新缓冲区，避免丢失未写入的日志

2. 本地文件中转后上传GCS #

先将日志写入本地临时文件，任务结束后通过Dataproc自带的GCS工具（如gsutil）或PySpark的Hadoop API将文件上传至GCS。

示例代码： #

import logging
import subprocess
import tempfile
import os

# 配置日志写入本地临时文件
temp_dir = tempfile.gettempdir()
local_log_path = os.path.join(temp_dir, "pyspark_task.log")
logging.basicConfig(
    filename=local_log_path,
    format="%(asctime)s %(message)s",
    filemode="w",
    level=logging.INFO
)

# 执行PySpark任务逻辑
logging.info("PySpark任务开始执行")
# ... 你的任务代码 ...

# 任务结束后上传日志到GCS
try:
    subprocess.run(
        ["gsutil", "cp", local_log_path, "gs://bucket_name/newfile.log"],
        check=True
    )
except subprocess.CalledProcessError as e:
    logging.error(f"上传日志到GCS失败: {str(e)}")
finally:
    # 清理本地临时文件
    if os.path.exists(local_log_path):
        os.remove(local_log_path)

注意事项： #

• 本地临时文件建议放在/tmp目录（Dataproc节点的临时存储，会自动清理），避免占用持久化磁盘空间
• 确保集群节点已安装gsutil（Dataproc默认预装）

3. 利用Dataproc原生日志集成（无需修改代码） #

如果不需要自定义日志格式，可直接配置Dataproc将YARN/Spark日志自动导出到GCS，无需修改任务代码：

• 提交作业时配置：
  提交PySpark作业时添加以下参数，将Spark事件日志和YARN日志导出到GCS：

  gcloud dataproc jobs submit pyspark \
      --cluster=your-cluster-name \
      --properties spark.eventLog.enabled=true,spark.eventLog.dir=gs://bucket_name/spark-logs \
      --properties yarn.log-aggregation-enable=true,yarn.nodemanager.remote-app-log-dir=gs://bucket_name/yarn-logs \
      your_script.py
  

• 集群全局配置：
  创建集群时配置日志聚合到GCS，后续所有作业日志都会自动导出：

  gcloud dataproc clusters create your-cluster-name \
      --region=your-region \
      --properties spark.eventLog.enabled=true,spark.eventLog.dir=gs://bucket_name/spark-logs \
      --properties yarn.log-aggregation-enable=true,yarn.nodemanager.remote-app-log-dir=gs://bucket_name/yarn-logs
  

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