java - Spark - 将 scala 转换为 java

Question

使用apache-spark处理数据。

鉴于这样scala代码:

val rdd1 = sc.cassandraTable("player", "playerinfo").select("key1", "value")
val rdd2 = rdd1.map(row => (row.getString("key1"), row.getLong("value")))

基本上，它隐藏了 RDD 'rdd1' 到另一个 RDD 'rdd2'，但它将'rdd1'存储为键值对形式。

注意源数据来自cassandra和keys1是复合键的一部分并且 value是值。

那么如何将其转换为Java这样我就会有一个 JavaPairRDD<String,Long>使用 Spark Java API？我已经有一个cassandraRowsRDD从下面的Java代码中成功生成:

  JavaRDD<String> cassandraRowsRDD = javaFunctions(sc).cassandraTable("player", "playerinfo")
            .map(new Function<CassandraRow, String>() {
                @Override
                public String call(CassandraRow cassandraRow) throws Exception {
                    return cassandraRow.toString();
                }
            });

Answer 1

CassandraJavaRDD 继承 mapToPair 方法。 Java中可以调用它来获取键值对RDD。

    JavaPairRDD<String, String> cassandraKeyValuePairs = javaFunctions(sc).cassandraTable("player", "playerinfo").mapToPair(
            new PairFunction<CassandraRow, String, String>() {
                @Override
                public Tuple2<String, String> call(CassandraRow row) throws Exception {
                    return new Tuple2(row.getString("key1"), row.getLong("value"));
                }
            }
    );

您还可以在 cassandraRowsRDD 上调用该函数。