如何将 spark 数据框模式存储在 databricks 笔记本的变量中？答案

【问题标题】：how to store spark dataframe schema in a variable in databricks notebook?如何将 spark 数据框模式存储在 databricks 笔记本的变量中？
【发布时间】：2019-10-11 09:28:55
【问题描述】：

使用 spark scala azure 数据块。

我有一个包含 100 多列的数据框（df1）。我需要创建另一个数据框（df2）并且也希望有相同的模式。如何将 df1 的模式存储到变量中并应用于 df2

val inputDF = Seq(("00163E0F765C1ED79593228BF70CEE41" ,"PD PUMPS")
                       ,("00164E0F775C1ED79593228BF70CEE42" ,"PD PUMPS")
                       ,("00165E0F785C1ED79593228BF70CEE43" ,"PD PUMPS")
                       ,("00166E0F795C1ED79593228BF70CEE44" ,"PD PUMPS")
                       ,("00167E0F405C1ED79593228BF70CEE45" ,"PD PUMPS")
                   ).toDF("objectID")
val expectedDF = Seq(("00163E0F765C1ED79593228BF70CEE41" ,"PD PUMPS1")
                       ,("00164E0F775C1ED79593228BF70CEE42" ,"PD PUMPS1")
                       ,("00165E0F785C1ED79593228BF70CEE43" ,"PD PUMPS1")
                       ,("00166E0F795C1ED79593228BF70CEE44" ,"PD PUMPS1")
                       ,("00167E0F405C1ED79593228BF70CEE45" ,"PD PUMPS1")
                       ).toDF("objectID","equipmentName", inputDF.schema)

这样做的目的是：我实际上是在编写单元测试用例。我有一个函数将一列添加到传递的数据框中（参数化）。所以我需要创建 inputDF，然后我需要在 inputDF 中再创建 1 列的 expectedDF。

【问题讨论】：

我尝试了 databricks 笔记本中没有的 df.schema 方法

标签： apache-spark azure-databricks

【解决方案1】：

鉴于对于 inputDF 生成，您应该使用仅包含一列的序列（或在 toDF 方法中传递两个字符串），我将执行以下操作：

val inputDF = Seq(("00163E0F765C1ED79593228BF70CEE41")
  ,("00164E0F775C1ED79593228BF70CEE42")
  ,("00165E0F785C1ED79593228BF70CEE43")
  ,("00166E0F795C1ED79593228BF70CEE44")
  ,("00167E0F405C1ED79593228BF70CEE45")
).toDF("objectID")

val seq = Seq(("00163E0F765C1ED79593228BF70CEE41" ,"PD PUMPS1")
  ,("00164E0F775C1ED79593228BF70CEE42" ,"PD PUMPS1")
  ,("00165E0F785C1ED79593228BF70CEE43" ,"PD PUMPS1")
  ,("00166E0F795C1ED79593228BF70CEE44" ,"PD PUMPS1")
  ,("00167E0F405C1ED79593228BF70CEE45" ,"PD PUMPS1")
)

val rdd = spark.sparkContext.parallelize(seq)

val rows: RDD[Row] = rdd.map((row: (String, String)) => {
  Row(row.productIterator.toList:_*)
})

val expectedDF = spark.createDataFrame(rows,
  inputDF.schema.add(StructField("NewColumn", org.apache.spark.sql.types.StringType )))

df.show()

这个想法是使用 createDataFrame 从序列中创建一个 Dataframe，并将旧数据帧的架构（即 StructType）传递给它，并添加一个条目（StructField）。

【讨论】：