如何在Spark的worker节点中读取文件？

Question

我正在开发一个spark应用程序，它需要读取工作节点上的文件。我为我的rdd创建了一个映射，如下所示：

        SparkConf conf = new SparkConf().setAppName("startingSpark").setMaster("local[*]");
        JavaSparkContext sc = new JavaSparkContext(conf);              
        JavaRDD<String> file = sc.textFile("D:\\myFile.txt");
        JavaPairRDD<String, String> pairs = file.mapToPair(s -> new Tuple2<>((int)(Integer.parseInt(s.split(",")[1])/100)+"", s.split(",")[1])/100)+"#"+s));

现在，我想根据每个worker节点上的键值读取文件。

JavaPairRDD<String, String> rdd1 = pairs.reduceByKey(new Function2<String, String, String>() {
                
        @Override
        public String call(String v1, String v2) throws Exception {
            String key=v2.split("#")[0];
            JavaRDD <String> br =sc.textFile(key+".txt");
                    return v1+" "+v2;
                }
            }); 

它会给出一个错误：

java.io.NotSerializableException: org.apache.spark.api.java.JavaSparkContext 

我尝试过用我的主类实现java.io.Serializable，但是没有成功。任何帮助都是非常感谢的。

Answer 1

java.io.NotSerializableException: org.apache.spark.api.java.JavaSparkContext 

上面的异常是因为您的所有工作节点都无法识别路径"D:\myFile.txt“

您不需要在代码中实现任何额外的逻辑，只需使用将源文件放入HDFS路径，然后替换为下面

JavaRDD<String> file = sc.textFile("hdfs:\\localhost\xx\\xx.txt");