scala - 如何在spark中将rdd对象转换为dataframe

Question

如何将 RDD (org.apache.spark.rdd.RDD[org.apache.spark.sql.Row]) 转换为 Dataframe org.apache.spark。 sql.DataFrame。我使用 .rdd 将数据帧转换为 rdd。处理后我希望它回到数据框中。我该如何做到这一点？

Answer 1

此代码在 Spark 2.x 和 Scala 2.11 中完美运行

导入必要的类

import org.apache.spark.sql.{Row, SparkSession}
import org.apache.spark.sql.types.{DoubleType, StringType, StructField, StructType}

创建SparkSession对象，这是 spark

val spark: SparkSession = SparkSession.builder.master("local").getOrCreate
val sc = spark.sparkContext // Just used to create test RDDs

让我们 RDD做到DataFrame

val rdd = sc.parallelize(
  Seq(
    ("first", Array(2.0, 1.0, 2.1, 5.4)),
    ("test", Array(1.5, 0.5, 0.9, 3.7)),
    ("choose", Array(8.0, 2.9, 9.1, 2.5))
  )
)

##方法1 使用SparkSession.createDataFrame(RDD obj) .

val dfWithoutSchema = spark.createDataFrame(rdd)

dfWithoutSchema.show()
+------+--------------------+
|    _1|                  _2|
+------+--------------------+
| first|[2.0, 1.0, 2.1, 5.4]|
|  test|[1.5, 0.5, 0.9, 3.7]|
|choose|[8.0, 2.9, 9.1, 2.5]|
+------+--------------------+

##方法2 使用SparkSession.createDataFrame(RDD obj)并指定列名称。

val dfWithSchema = spark.createDataFrame(rdd).toDF("id", "vals")

dfWithSchema.show()
+------+--------------------+
|    id|                vals|
+------+--------------------+
| first|[2.0, 1.0, 2.1, 5.4]|
|  test|[1.5, 0.5, 0.9, 3.7]|
|choose|[8.0, 2.9, 9.1, 2.5]|
+------+--------------------+

##方法3(问题的实际答案) 这种方式需要输入rdd类型应为 RDD[Row] .

val rowsRdd: RDD[Row] = sc.parallelize(
  Seq(
    Row("first", 2.0, 7.0),
    Row("second", 3.5, 2.5),
    Row("third", 7.0, 5.9)
  )
)

创建架构

val schema = new StructType()
  .add(StructField("id", StringType, true))
  .add(StructField("val1", DoubleType, true))
  .add(StructField("val2", DoubleType, true))

现在应用 rowsRdd和schema至createDataFrame()

val df = spark.createDataFrame(rowsRdd, schema)

df.show() 
+------+----+----+
|    id|val1|val2|
+------+----+----+
| first| 2.0| 7.0|
|second| 3.5| 2.5|
| third| 7.0| 5.9|
+------+----+----+