java - Spark - 在字符串匹配后提取行并将其保存在 ArrayList 中

Question

我是 Spark 新手，尝试提取包含“主题:”的行并将其保存在数组列表中。我没有遇到任何错误，但数组列表为空。你能指导我哪里出错了吗？或者最好的方法来做到这一点？

import java.util.*;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.VoidFunction;
public final class extractSubject {

public static void main(String[] args) {



    SparkConf sparkConf = new SparkConf().setMaster("local[1]").setAppName("JavaBookExample");
    JavaSparkContext sc = new JavaSparkContext(sparkConf);

    JavaRDD<String> sample = sc.textFile("/Users/Desktop/sample.txt");
    final ArrayList<String> list = new ArrayList<>();
    sample.foreach(new VoidFunction<String>(){

                    public void call(String line) {

                       if (line.contains("Subject:")) {
                           System.out.println(line);
                           list.add(line);
                       }
                   }}
    );

    System.out.println(list);
    sc.stop();
}
 }

Answer 1

请记住，Spark 应用程序是分布式且并行运行的。因此，您无法修改 Spark 执行的函数之外的变量。

相反，您需要从这些函数返回结果。在您的情况下，您需要 flatMap (而不是没有结果的 foreach)，它连接作为函数结果返回的集合。

如果某行匹配，则返回包含匹配行的列表，否则返回空列表。

要在 main 函数中打印数据，您首先必须通过调用 collect() 来收集主节点中可能分布的数据。

这里是一个例子:

import java.util.*;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;

public final class extractSubject {

    public static void main(String[] args) {
        SparkConf sparkConf = new SparkConf().setMaster("local[1]").setAppName("JavaBookExample");
        JavaSparkContext sc = new JavaSparkContext(sparkConf);

        //JavaRDD<String> sample = sc.textFile("/Users/Desktop/sample.txt");
        JavaRDD<String> sample = sc.parallelize(Arrays.asList("Subject: first",
                                                              "nothing here",
                                                              "Subject: second",
                                                              "dummy"));

        JavaRDD<String> subjectLinesRdd = sample.flatMap(new FlatMapFunction<String, String>() {
            public Iterable<String> call(String line) {
                if (line.contains("Subject:")) {
                    return Collections.singletonList(line);  // line matches → return list with the line as its only element
                } else {
                    return Collections.emptyList();  // ignore line → return empty list
                }
            }
        });

        List<String> subjectLines = subjectLinesRdd.collect();  // collect values from Spark workers
        System.out.println(subjectLines);  // → "[Subject: first, Subject: second]"

        sc.stop();
    }
}