我尝试在 spark 任务中使用 lambda 表达式，它抛出“java.lang.IllegalArgumentException:无效的 lambda 反序列化”异常。当代码类似于 "transform(pRDD->pRDD.map(t->t._2))"时会抛出此异常。代码片段如下。

JavaPairDStream<String,Integer> aggregate = pairRDD.reduceByKey((x,y)->x+y);
JavaDStream<Integer> con = aggregate.transform(
(Function<JavaPairRDD<String,Integer>, JavaRDD<Integer>>)pRDD-> pRDD.map( 
(Function<Tuple2<String,Integer>,Integer>)t->t._2));


JavaPairDStream<String,Integer> aggregate = pairRDD.reduceByKey((x,y)->x+y);
JavaDStream<Integer> con = aggregate.transform(
(Function<JavaPairRDD<String,Integer>, JavaRDD<Integer>> & Serializable)pRDD-> pRDD.map( 
(Function<Tuple2<String,Integer>,Integer> & Serializable)t->t._2));

以上两个选项均无效。好像我在下面传递对象“f”作为参数而不是 lambda 表达式“t->t_.2”。它有效。

Function f = new Function<Tuple2<String,Integer>,Integer>(){
@Override
public Integer call(Tuple2<String,Integer> paramT1) throws Exception {
return paramT1._2;
}
}; 

我可以知道用 lambda 表达式表达函数的正确格式是什么吗？

    public static void main(String[] args) {

            Function f = new Function<Tuple2<String,Integer>,Integer>(){

                @Override
                public Integer call(Tuple2<String,Integer> paramT1) throws Exception {
                    return paramT1._2;
                }

            };

            JavaStreamingContext ssc = JavaStreamingFactory.getInstance();

            JavaReceiverInputDStream<String> lines = ssc.socketTextStream("localhost", 9999);
            JavaDStream<String> words =  lines.flatMap(s->{return Arrays.asList(s.split(" "));});
            JavaPairDStream<String,Integer> pairRDD =  words.mapToPair(x->new Tuple2<String,Integer>(x,1));
            JavaPairDStream<String,Integer> aggregate = pairRDD.reduceByKey((x,y)->x+y);
            JavaDStream<Integer> con = aggregate.transform(
                    (Function<JavaPairRDD<String,Integer>, JavaRDD<Integer>>)pRDD-> pRDD.map( 
                            (Function<Tuple2<String,Integer>,Integer>)t->t._2));
          //JavaDStream<Integer> con = aggregate.transform(pRDD-> pRDD.map(f)); It works
            con.print();

            ssc.start();
            ssc.awaitTermination();


        }

最佳答案

我不知道为什么 lambda 不起作用。也许问题在于嵌套在 lambda 中的 lambda。 Spark 文档似乎承认了这一点。

对比 http://spark.apache.org/docs/latest/programming-guide.html#basics 中的示例:

JavaRDD<String> lines = sc.textFile("data.txt");
JavaRDD<Integer> lineLengths = lines.map(s -> s.length());
int totalLength = lineLengths.reduce((a, b) -> a + b);

以 http://spark.apache.org/docs/latest/streaming-programming-guide.html#transform-operation 中的例子为例:

import org.apache.spark.streaming.api.java.*;
// RDD containing spam information
final JavaPairRDD<String, Double> spamInfoRDD = jssc.sparkContext().newAPIHadoopRDD(...);

JavaPairDStream<String, Integer> cleanedDStream = wordCounts.transform(
  new Function<JavaPairRDD<String, Integer>, JavaPairRDD<String, Integer>>() {
    @Override public JavaPairRDD<String, Integer> call(JavaPairRDD<String, Integer> rdd) throws Exception {
      rdd.join(spamInfoRDD).filter(...); // join data stream with spam information to do data cleaning
      ...
    }
  });

第二个示例使用 Function 子类而不是 lambda，可能是因为您发现了相同的问题。

我不知道这对你是否有用，但嵌套的 lambda 在 Scala 中确实有效。考虑前面示例的 Scala 版本:

val spamInfoRDD = ssc.sparkContext.newAPIHadoopRDD(...) // RDD containing spam information

val cleanedDStream = wordCounts.transform(rdd => {
  rdd.join(spamInfoRDD).filter(...) // join data stream with spam information to do data cleaning
  ...
})

关于java - Apache Spark Lambda 表达式 - 序列化问题，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/31697881/