求助：本地使用Spark Runner运行Apache Beam WordCount示例失败

嘿，我帮不少开发者踩过Spark Runner本地跑Beam示例的坑，咱们一步步拆解问题、搞定它：

1. 先检查依赖是否正确配置 #

Spark Runner需要额外的依赖包，而且Beam和Spark的版本必须兼容（比如Beam 2.50+对应Spark 3.x）。如果用Maven，你需要在pom.xml里添加这些runtime依赖：

<dependency>
  <groupId>org.apache.beam</groupId>
  <artifactId>beam-runners-spark-3</artifactId>
  <version>你的Beam版本，比如2.54.0</version>
  <scope>runtime</scope>
</dependency>
<dependency>
  <groupId>org.apache.spark</groupId>
  <artifactId>spark-core_2.12</artifactId>
  <version>3.3.4</version>
  <scope>runtime</scope>
</dependency>

如果是Gradle，对应调整依赖配置即可。

2. 必须明确指定Spark Runner和本地模式 #

默认情况下Beam用的是DirectRunner，你得手动切换到SparkRunner，还要设置本地Spark的master地址，不然程序会尝试连接远程Spark集群（当然连不上）。修改你的Pipeline配置代码：

PipelineOptions options = PipelineOptionsFactory.create();
// 指定Spark Runner
options.setRunner(SparkRunner.class);
// 转换为Spark专属配置，设置本地模式（用所有可用核心）
SparkPipelineOptions sparkOptions = options.as(SparkPipelineOptions.class);
sparkOptions.setSparkMaster("local[*]");
// 可选：如果内存不够，增加Spark驱动和执行器的内存
sparkOptions.setSparkConf("spark.driver.memory", "2g");
sparkOptions.setSparkConf("spark.executor.memory", "2g");

Pipeline p = Pipeline.create(options);

3. 确认输入文件路径和权限 #

你用的file:///tmp/shakespeare.txt要确保：

• 文件确实存在于本地/tmp目录下（Windows系统要改成file:///C:/tmp/shakespeare.txt这类格式）
• 运行程序的用户有读取这个文件的权限
• 如果是Windows，注意路径里的斜杠方向，别写错

4. 补全DoFn实现和完整的Pipeline逻辑 #

你的代码里processElement方法没写完，而且MinimalWordCount示例还需要后续的计数和输出步骤，不然管道不完整也会报错。补全后的完整核心逻辑大概是这样：

p.apply(TextIO.read().from("file:///tmp/shakespeare.txt"))
 .apply("ExtractWords", ParDo.of(new DoFn<String, String>() {
     @ProcessElement
     public void processElement(ProcessContext c) {
         String line = c.element();
         // 按非单词字符分割，提取有效单词
         for (String word : line.split("\\W+")) {
             if (!word.isEmpty()) {
                 c.output(word);
             }
         }
     }
 }))
 .apply(Count.perElement())
 // 把KV格式转成可读的字符串
 .apply(MapElements.via(new SimpleFunction<KV<String, Long>, String>() {
     @Override
     public String apply(KV<String, Long> input) {
         return input.getKey() + ": " + input.getValue();
     }
 }))
 // 输出结果到指定目录
 .apply(TextIO.write().to("file:///tmp/wordcounts"));

// 别忘了启动管道
p.run().waitUntilFinish();

5. 排查日志找具体错误 #

如果还是失败，一定要看程序输出的日志，Spark Runner会抛出具体的异常（比如依赖缺失、路径不存在、内存不足），根据日志提示针对性解决会更快。

内容的提问来源于stack exchange，提问作者harshvardhan.agr