java - 无法在 Spark 中联合两个 CassandraJavaRDD

Question

由于从 Cassandra 查询数据有限制，我尝试使用 Spark 批量读取数据并将其存储在 RDD 中。

然后我使用 union 函数添加所有 RDD 。

这是我的代码。

private void getDataFromCassandra(JavaSparkContext sc) {


    CassandraJavaRDD<CassandraRow> cassandraRDD = null ;
    CassandraJavaRDD<CassandraRow> cassandraRDD2  = null;

    While(Some Condition)

     cassandraRDD = CassandraJavaUtil
                .javaFunctions(sc).cassandraTable("dmp", "table").select("abc", "xyz")
                .where("pid IN ('" + sb + "')");

    if(cassandraRDD2==null){


     cassandraRDD2=cassandraRDD;
    }
    else{
        cassandraRDD2 =  cassandraRDD2.union(cassandraRDD);
    }
}             

}

但是在联合中我收到以下错误。

类型不匹配:无法从 JavaRDD 转换为 CassandraJavaRDD

尽管这两个 RDD 的类型相似。

所以 1) 我应该将 Cast 应用为

 cassandraRDD2 =  (CassandraJavaRDD<CassandraRow>) cassandraRDD2.union(cassandraRDD);

2) 或者将其中一个 RDD 的类型更改为 JavaRDD

Answer 1

出现此问题的原因是根据 docs :

Method: union(JavaRDD other) Return the union of this RDD and another one.

Return Value: JavaRDD

因此不匹配。

因为根据this :

public class CassandraJavaRDD<R> extends JavaRDD<R> {
...
}

CassandraJavaRDD 类扩展了 JavaRDD，因此您可以使用:

JavaRDD<CassandraRow> cassandraRDD = null;
JavaRDD<CassandraRow> cassandraRDD2 = null;

因此 union() 方法的返回值将与其类型匹配。