如何配置Apache Beam Pipeline使用GCS模拟器作为temp_location

1. 确保依赖版本兼容 #

使用Apache Beam SDK 2.30.0及以上版本（DataFlow基于Beam），早期版本对GCS模拟器的支持不完善，可能会忽略STORAGE_EMULATOR_HOST环境变量。

2. 完整设置环境变量 #

启动Pipeline前，在终端中设置以下环境变量（确保覆盖所有可能的配置入口）：

export STORAGE_EMULATOR_HOST=http://localhost:8080
export CLOUDSDK_STORAGE_EMULATOR_HOST=http://localhost:8080
export GOOGLE_CLOUD_PROJECT=dummy-project  # 任意非空项目ID即可
export CLOUDSDK_CORE_PROJECT=dummy-project
export NO_PROXY=localhost,127.0.0.1  # 避免公司代理拦截本地模拟器请求

3. 提前创建模拟器中的存储桶 #

使用gsutil操作模拟器创建桶（需先设置上述环境变量）：

gsutil mb gs://your-emulator-bucket

4. 代码层面强制绑定模拟器（针对不同语言） #

Python示例 #

在初始化Pipeline前，显式指定GCS IO使用模拟器地址：

from apache_beam.options.pipeline_options import PipelineOptions, GoogleCloudOptions
from apache_beam.io.gcp.gcsio import GcsIO

# 基础配置
options = PipelineOptions()
gcp_options = options.view_as(GoogleCloudOptions)
gcp_options.project = "dummy-project"
gcp_options.temp_location = "gs://your-emulator-bucket/temp"

# 强制绑定模拟器GCS客户端
GcsIO.set_default_instance(GcsIO(
    endpoint_url="http://localhost:8080",
    project_id="dummy-project"
))

# 后续启动Pipeline逻辑
pipeline = beam.Pipeline(options=options)
# ... 你的Pipeline代码

Java示例 #

通过GcsOptions强制指定模拟器的GCS工具类：

import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.extensions.gcp.storage.GcsOptions;
import org.apache.beam.sdk.extensions.gcp.storage.emulator.EmulatorGcsUtilFactory;

public class YourDataFlowJob {
    public static void main(String[] args) {
        PipelineOptions options = PipelineOptionsFactory.fromArgs(args).create();
        GcsOptions gcsOptions = options.as(GcsOptions.class);
        
        // 绑定模拟器GCS工具
        gcsOptions.setGcsUtilFactoryClass(EmulatorGcsUtilFactory.class);
        gcsOptions.setTempLocation("gs://your-emulator-bucket/temp");
        gcsOptions.setProject("dummy-project");
        
        // 启动Pipeline
        // ... 你的Pipeline代码
    }
}

5. 验证模拟器连接 #

可以用gsutil ls gs://your-emulator-bucket测试是否能正常访问模拟器桶，确认没问题后再启动DataFlow Pipeline。

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