10

我有一个包含键值对的 JavaPairDStream。我需要将它转换为 HashMap。我已经尝试通过在其上调用“collectAsMap()”函数及其工作来对普通 JavaPairRDD 执行相同操作，但是当我尝试在 DStream 上执行相同操作时，它失败了。

我试图通过使用“foreachRDD”函数将“JavaPairDStream”转换为“JavaPairRDD”，然后在 JavaPairRDD 上使用“collectAsMap()”函数来实现相同的目的。

Map<String,String> value= new HashMap<String,String>();
            value=line.collectAsMap();

//Here "line" is a "JavaPairRDD<String,String>".

它没有给出任何编译错误，但是当我运行程序时它会失败并抛出如下错误。

java.lang.ClassCastException: [Ljava.lang.Object; cannot be cast to [Lscala.Tuple2;
    at org.apache.spark.rdd.PairRDDFunctions.collectAsMap(PairRDDFunctions.scala:447)
    at org.apache.spark.api.java.JavaPairRDD.collectAsMap(JavaPairRDD.scala:464)
    at attempt1.CSV_Spark$3.call(CSV_Spark.java:109)
    at attempt1.CSV_Spark$3.call(CSV_Spark.java:1)

我不确定我的方法是否正确。普通的“JavaPairRDD”和由“foreachRDD”函数创建的有什么区别吗？为什么相同的方法适用于普通的“JavaPairRDD”，但当我将其应用于通过在 JavaPairDStream 上应用“foreachRDD”函数创建的“JavaPairRDD”时会失败。如果我在任何地方出错，请告诉我。另外，如果还有其他方法，请在此处发布。谢谢。

4

2 回答 2

1

在编译时，接受向下转换，因为Map和HashMap都在相同的继承中。虽然我们没有得到任何编译时错误，但我们会在运行时得到 ClassCastException。为了避免这个问题，你可以试试这个：

代码：

JavaPairRDD<K, V> javaRDDPair  = rddInstance.mapToPair(new PairFunction<T, K, V>() {
   @Override
    public Tuple2<K, V> call(final T value) {
    // statements 
    // operations on value
    return new Tuple2<K, V>(KTypeValue, VTypeValue);
    }
    });

    Map<K,V> map =  javaRDDPair.collectAsMap();
    HashMap<K,V> hmap = new HashMap<K,V>(map);

注意：rddInstance是JavaRDD类型的对象。

假设我们有一个JavaRDD，其中包含T 类型的值。在对其进行转换后，我们创建了包含 < K,V >对的JavaPairRDD。现在的要求是将 JavaPairRDD 转换为 HashMap 对象，以便在您的应用程序中进行进一步计算。使用collectAsMap方法并将其结果分配给Map对象本身。之后，您可以通过传递Map实例来创建HashMap。

于 2016-12-01T07:18:54.643 回答
0

你可以试试

JavaPairDStream stream =... 
JavaPairRDD pairRdd=stream.compute(validTime);

这相当于一种围绕 Time 类型的 validTime 瞬间的分桶，因为您正在对流进行推理。

或者，使用 forEachRDD，然后换行

JavaPairRDD<K,V> wrapRDD(RDD<scala.Tuple2<K,V>> rdd)

从那里收集AsMap。

java.util.Map<K,V>  collectAsMap()
于 2015-10-15T12:23:28.940 回答