java - 使用java的RDD函数中的类型不匹配

Question

我是 Spark 新手，我编写了一些简单的代码，但在执行任务时遇到了问题。我正在使用 Java 1.7 SE 和 Spark 2.0.1 来执行一些任务。

下面是我的代码。

JavaRDD<String> newLine = line.flatMap(new FlatMapFunction<String, String>() {
    public List<String> call(String s) throws Exception {
        List<String> ret = new ArrayList<String>();
        List<String> ls = Arrays.asList(s.split("\t"));

        String values = ls.get(ls.size()-1);
        List<String> value = Arrays.asList(values.split("\\|"));

        for(int i=0;i<value.size();++i){
            String ns = ls.get(0)+"\t"+ls.get(1)+"\t"+ls.get(2)+"\t"+ls.get(3)+"\t"+ls.get(4)+"\t"+ls.get(5);
            ns = ns + "\t" + value.get(i);
            ret.add(ns);
        }
        return ret;
    }
});

错误发生在 public List<String> call(String s) throws Exception因为List<String> 。错误消息显示我尝试尝试使用不兼容的返回类型。我理解了这条消息，但我找不到正确的解决方案。另外，当我使用maven repo将spark库从1.6.0更改为2.0.1时，就发生了这个错误。如果您能为我提供有关如何解决此问题的指导，我将不胜感激。

为了方便起见，我附上了完整的代码，如下。

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Iterator;
import java.util.List;
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 class DataTrans {
    public static void main(String[] args) throws Exception{
        String inputFile = args[0];
        String outputFile = args[1];

        SparkConf conf = new SparkConf().setAppName("Data Transformation")
                .set("spark.serializer","org.apache.spark.serializer.KryoSerializer").setMaster("local[*]");

        JavaSparkContext sc = new JavaSparkContext(conf);

        /* Load our input data. */
        JavaRDD<String> input = sc.textFile(inputFile);

        JavaRDD<String> line = input.flatMap(new FlatMapFunction<String, String>() {
            public Iterator<String> call(String s) throws Exception {
                return Arrays.asList(s.split("\n")).iterator();
            }
        });

        JavaRDD<String> newLine = line.flatMap(new FlatMapFunction<String, String>() {
            public List<String> call(String s) throws Exception {
                List<String> ret = new ArrayList<String>();
                List<String> ls = Arrays.asList(s.split("\t"));

                String values = ls.get(ls.size()-1);
                List<String> value = Arrays.asList(values.split("\\|"));

                for(int i=0;i<value.size();++i){
                    String ns = ls.get(0)+"\t"+ls.get(1)+"\t"+ls.get(2)+"\t"+ls.get(3)+"\t"+ls.get(4)+"\t"+ls.get(5);
                    ns = ns + "\t" + value.get(i);
                    ret.add(ns);
                }
                return ret;
            }
        });
        newLine.saveAsTextFile(outputFile);

        sc.stop();
        sc.close();
    }
} 

谢谢

Answer 1

它应该返回 Iterator<String>不是List<String>在 call() 返回类型中

   JavaRDD<String> newLine = line.flatMap(new FlatMapFunction<String, String>() {
        public Iterator<String> call(String s) throws Exception {
            List<String> ret = new ArrayList<String>();
            List<String> ls = Arrays.asList(s.split("\t"));

            String values = ls.get(ls.size()-1);
            List<String> value = Arrays.asList(values.split("\\|"));

            for(int i=0;i<value.size();++i){
                String ns = ls.get(0)+"\t"+ls.get(1)+"\t"+ls.get(2)+"\t"+ls.get(3)+"\t"+ls.get(4)+"\t"+ls.get(5);
                ns = ns + "\t" + value.get(i);
                ret.add(ns);
            }
            return ret.iterator();
        }
    });