将架构从一个数据帧复制到另一个数据帧

Question

我正在尝试将现有数据帧的架构更改为另一个数据帧的架构。

数据帧 1：

Column A | Column B | Column C | Column D
   "a"   |    1     |   2.0    |   300
   "b"   |    2     |   3.0    |   400
   "c"   |    3     |   4.0    |   500

数据帧 2：

Column K | Column B | Column F
   "c"   |    4     |   5.0
   "b"   |    5     |   6.0
   "f"   |    6     |   7.0

所以我想在第二个数据帧上应用第一个数据帧的架构。因此，所有相同的列都保持不变。数据帧 2 中不在 1 中的列将被删除。其他变为"空"。

输出

Column A | Column B | Column C | Column D
 "NULL"  |    4     |   "NULL" |  "NULL"
 "NULL"  |    5     |   "NULL" |  "NULL"
 "NULL"  |    6     |   "NULL" |  "NULL"

所以我提出了一个可能的解决方案：

val schema = df1.schema
val newRows: RDD[Row] = df2.map(row => {
  val values = row.schema.fields.map(s => {
    if(schema.fields.contains(s)){
      row.getAs(s.name).toString
    }else{
      "NULL"
    }
  })
  Row.fromSeq(values)
})
sqlContext.createDataFrame(newRows, schema)}

现在如您所见，这将不起作用，因为架构包含字符串、Int 和双精度。我所有的行都有字符串值。

这就是我卡住的地方，有没有办法自动将我的值类型转换为架构？

Answer 1

如果模式是平面的，我将简单地映射每个现有模式并select必需的列：

val exprs = df1.schema.fields.map { f => 
  if (df2.schema.fields.contains(f)) col(f.name)
  else lit(null).cast(f.dataType).alias(f.name) 
}
df2.select(exprs: _*).printSchema
// root
//  |-- A: string (nullable = true)
//  |-- B: integer (nullable = false)
//  |-- C: double (nullable = true)
//  |-- D: integer (nullable = true)

Answer 2

在 2018 年工作 （Spark 2.3） 读取 .sas7bdat

斯卡拉

val sasFile = "file.sas7bdat"
val dfSas = spark.sqlContext.sasFile(sasFile)
val myManualSchema = dfSas.schema //getting the schema from another dataframe
val df = spark.read.format("csv").option("header","true").schema(myManualSchema).load(csvFile)

PD：spark.sqlContext.sasFile 使用 saurfang 库，您可以跳过该部分代码并从另一个数据帧获取架构。

Answer 3

以下是实现相同目标的简单 PYSPARK 步骤：

df = <dataframe whose schema needs to be copied>
df_tmp = <dataframe with result with fewer fields> 
#Note: field names from df_tmp must match with field names from df
df_tmp_cols = [colmn.lower() for colmn in df_tmp.columns]
for col_dtls in df.dtypes:
  col_name, dtype = col_dtls
  if col_name.lower() in df_tmp_cols:
    df_tmp = df_tmp.withColumn(col_name,f.col(col_name).cast(dtype))
  else:
    df_tmp = df_tmp.withColumn(col_name,f.lit(None).cast(dtype)) 
df_fin = df_tmp.select(df.columns) #Final dataframe

Answer 4

您可以使用如下查询在数据帧上执行左联接：-

SELECT Column A, Column B, Column C, Column D FROM foo LEFT JOIN BAR ON Column C = Column C

请在这篇文章中@zero323查看答案：-

Spark 为数据帧联接指定多个列条件

谢谢查尔斯。