将 JavaRDD<Tuple2<Object, long[]>> 转换为 Java 中的 Spark Dataset<Row>答案

【问题标题】：Convert a JavaRDD<Tuple2<Object, long[]>> into a Spark Dataset<Row> in Java将 JavaRDD<Tuple2<Object, long[]>> 转换为 Java 中的 Spark Dataset<Row>
【发布时间】：2021-04-12 08:59:47
【问题描述】：

在 Java（不是 Scala！）中，Spark 3.0.1 有一个 JavaRDD 实例对象 neighborIdsRDD，其类型为 JavaRDD<Tuple2<Object, long[]>>。

我与生成JavaRDD相关的部分代码如下：

GraphOps<String, String> graphOps = new GraphOps<>(graph, stringTag, stringTag);
JavaRDD<Tuple2<Object, long[]>> neighborIdsRDD = graphOps.collectNeighborIds(EdgeDirection.Either()).toJavaRDD();

我不得不使用toJavaRDD() 获取JavaRDD，因为collectNeighborIds 返回一个org.apache.spark.graphx.VertexRDD<long[]> 对象(VertexRDD doc)。

但是，在我的应用程序的其余部分中，我需要从 collectNeighborIds 对象构建一个 Spark Dataset<Row>。

将JavaRDD<Tuple2<Object, long[]>> 转换为Dataset<Row> 的可能性和最佳方法是什么？

来自 cmets 的更新：

我根据 cmets 调整了代码：

        GraphOps<String, String> graphOps = new GraphOps<>(graph, stringTag, stringTag);
        JavaRDD<Tuple2<Object, long[]>> neighborIdsRDD = graphOps.collectNeighborIds(EdgeDirection.Either()).toJavaRDD();
        System.out.println("VertexRDD neighborIdsRDD is:");
        for (int i = 0; i < neighborIdsRDD.collect().size(); i++) {
            System.out.println(
                    ((Tuple2<Object, long[]>) neighborIdsRDD.collect().get(i))._1() + " -- " +
                            Arrays.toString(((Tuple2<Object, long[]>) neighborIdsRDD.collect().get(i))._2())
            );
        }

        Dataset<Row> dr = spark_session.createDataFrame(neighborIdsRDD.rdd(), Tuple2.class);
        System.out.println("converted Dataset<Row> is:");
        dr.show();

但我得到一个空数据集，如下所示：

VertexRDD neighborIdsRDD is:
4 -- [3]
1 -- [2, 3]
5 -- [3, 2]
2 -- [1, 3, 5]
3 -- [1, 2, 5, 4]
converted Dataset<Row> is:
++
||
++
||
||
||
||
||
++

【问题讨论】：

这能回答你的问题吗？ convert RDD to Dataset in Java Spark
我已经看到了这个问题，但是由于我在 Spark 和 Java 中很陌生，我不是专家，所以我很难理解如何为我的案例编写“bean”：在问题的情况下，他们“bean”了一个 Person.class，我怎么能为
@kitsune 你的 javardd 包含 Tuple2。试试sparkSession.createDataFrame(neighborIdsRDD.rdd(), Tuple2.class)
谢谢！我尝试了您的建议，但是当我尝试 .show() 时，我得到了一个空的数据集。我更新了这次尝试的问题。

标签： java apache-spark dataset spark-graphx

【解决方案1】：

我遇到了同样的情况，但幸运的是，我找到了取回 Dataframe 的解决方案。

解决方案代码在步骤[1]、[2]和[3]处注释。

GraphOps<String, String> graphOps = new GraphOps<>(graph, stringTag, stringTag);
System.out.println("VertexRDD neighborIdsRDD is:");
JavaRDD<Tuple2<Object, long[]>> neighborIdsRDD = graphOps.collectNeighborIds(EdgeDirection.Either()).toJavaRDD();
for (int i = 0; i < neighborIdsRDD.collect().size(); i++) {
    System.out.println(
            ((Tuple2<Object, long[]>) neighborIdsRDD.collect().get(i))._1() + " -- " +
                    Arrays.toString(((Tuple2<Object, long[]>) neighborIdsRDD.collect().get(i))._2())
    );
}

// [1] Define encoding schema
StructType graphStruct =  new StructType(new StructField[]{
        new StructField("father", DataTypes.LongType, false, Metadata.empty()),
        new StructField("children", DataTypes.createArrayType(DataTypes.LongType), false, Metadata.empty()),
});

// [2] Build a JavaRDD<Row> from a JavaRDD<Tuple2<Object,long[]>>
JavaRDD<Row> dr = neighborIdsRDD.map(tupla -> RowFactory.create(tupla._1(), tupla._2()));
        
// [3] Finally build the reqired Dataframe<Row>
Dataset<Row> dsr = spark_session.createDataFrame(dr.rdd(), graphStruct);

System.out.println("DATASET IS:");
dsr.show();

打印输出：

VertexRDD neighborIdsRDD is:
4 -- [3]
1 -- [2, 3]
5 -- [3, 2]
2 -- [1, 3, 5]
3 -- [1, 2, 5, 4]
DATASET IS:
+------+------------+
|father|    children|
+------+------------+
|     4|         [3]|
|     1|      [2, 3]|
|     5|      [3, 2]|
|     2|   [1, 3, 5]|
|     3|[1, 2, 5, 4]|
+------+------------+

【讨论】：