转换JavaRDD< Tuple2< Object,long []>进入Spark Dataset< Row>在Java中 [英] Convert a JavaRDD<Tuple2<Object, long[]>> into a Spark Dataset<Row> in Java
问题描述
在Java(不是Scala!)中,Spark 3.0.1具有
In Java (not Scala!) Spark 3.0.1 have a JavaRDD instance object neighborIdsRDD
which its type is JavaRDD<Tuple2<Object, long[]>>
.
与JavaRDD生成相关的部分代码如下:
Part of my code related to the generation of the JavaRDD is the following:
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文档).
I have had to get a JavaRDD using toJavaRDD()
because collectNeighborIds
returns a org.apache.spark.graphx.VertexRDD<long[]>
object (VertexRDD doc).
但是,在我的其余应用程序中,我需要有一个 Dataset< Row>
由 collectNeighborIds
对象构建.
However, in the rest of my application I need to have a Spark Dataset<Row>
built from the collectNeighborIds
object.
获得 JavaRDD< Tuple2< Object,long []>> 被转换为我根据注释调整了代码:
I adjusted the code basing from comments:
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();
但是我得到一个空的数据集,如下所示:
but I get an empty Dataset as follows:
VertexRDD neighborIdsRDD is:
4 -- [3]
1 -- [2, 3]
5 -- [3, 2]
2 -- [1, 3, 5]
3 -- [1, 2, 5, 4]
converted Dataset<Row> is:
++
||
++
||
||
||
||
||
++
推荐答案
我也遇到了同样的情况,但是幸运的是,我找到了一种解决方法,可以找回 Dataframe .
I was in your same situation, but fortunately I found a solution to get back a Dataframe.
在步骤 [1]
, [2]
和 [3]
处注释解决方案代码.
Solution code is commented at steps [1]
, [2]
and [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]|
+------+------------+
这篇关于转换JavaRDD< Tuple2< Object,long []>进入Spark Dataset< Row>在Java中的文章就介绍到这了,希望我们推荐的答案对大家有所帮助,也希望大家多多支持IT屋!