如何在Beam Python（Cloud Dataflow）中保存C++二进制日志？

核心方案：本地临时文件中转后上传GCS #

这是最直接且低复杂度的方案，避开GFile不支持fileno的问题，利用本地文件作为中转：

• 步骤1：在Python的DoFn里创建临时目录，通过环境变量或命令行参数指定C程序的glog日志输出到该目录（比如设置GLOG_log_dir环境变量，或者启动C程序时加--log_dir=临时目录路径参数）。
• 步骤2：启动C++进程时，将stdout和stderr直接重定向到临时目录下的单独文件（用Python的subprocess模块，指定stdout和stderr为本地文件句柄）。
• 步骤3：等C++进程执行完成后，读取临时目录里的所有日志文件（包括stdout、stderr、glog生成的.log文件）。
• 步骤4：用Beam的文件IO或者GCS客户端把这些文件上传到目标GCS路径，注意给每个日志文件加唯一标识（比如元素ID、时间戳）防止覆盖。
• 步骤5：清理临时文件，避免占用Worker节点磁盘空间。

示例代码片段：

import tempfile
import subprocess
import os
from google.cloud import storage
import apache_beam as beam

class RunCppBinary(beam.DoFn):
    def process(self, element):
        # 用临时目录自动管理文件清理
        with tempfile.TemporaryDirectory() as temp_dir:
            # 配置glog输出到临时目录
            os.environ['GLOG_log_dir'] = temp_dir
            # 生成带唯一标识的输出文件名，避免冲突
            elem_id = element.get('id', str(os.getpid()))
            stdout_file = os.path.join(temp_dir, f"stdout_{elem_id}.log")
            stderr_file = os.path.join(temp_dir, f"stderr_{elem_id}.log")
            
            # 启动C++二进制，重定向标准输出/错误
            with open(stdout_file, 'w') as out_f, open(stderr_file, 'w') as err_f:
                subprocess.run(
                    ["/path/to/your/cpp_binary", "--your-arguments"],
                    stdout=out_f,
                    stderr=err_f,
                    check=True
                )
            
            # 上传所有日志到GCS
            gcs_bucket = "your-target-bucket"
            client = storage.Client()
            bucket = client.bucket(gcs_bucket)
            
            # 上传stdout
            bucket.blob(f"logs/stdout_{elem_id}.log").upload_from_filename(stdout_file)
            # 上传stderr
            bucket.blob(f"logs/stderr_{elem_id}.log").upload_from_filename(stderr_file)
            # 上传glog生成的日志文件
            for file_name in os.listdir(temp_dir):
                if file_name.endswith(".log") and file_name not in [os.path.basename(stdout_file), os.path.basename(stderr_file)]:
                    local_path = os.path.join(temp_dir, file_name)
                    gcs_path = f"logs/glog_{elem_id}_{file_name}"
                    bucket.blob(gcs_path).upload_from_filename(local_path)

备选方案：合并glog与标准输出后捕获 #

如果不需要严格区分日志类型，可以修改C++程序的glog配置，让glog日志直接输出到stdout或stderr（设置FLAGS_logtostderr=1让日志只输出到stderr，或者FLAGS_alsologtostderr=1同时输出到文件和stderr）。这样Python侧可以用subprocess.PIPE直接捕获所有输出，再写入GCS。但这种方式会把业务输出和日志混在一起，不利于后续核查，适合对日志分类要求低的场景。

不推荐方案：dup2重定向 #

用dup2修改文件描述符的方式实现重定向，需要把GCS的写入转换成本地管道或临时文件，不仅实现复杂，还容易出现缓冲区同步问题，Worker节点故障时可能丢失日志，开发维护成本高，收益低，不建议采用。

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