Spark Dataframe基于动态选择的列提取列答案

【问题标题】：Spark Dataframe extracting columns based dynamically selected columnsSpark Dataframe基于动态选择的列提取列
【发布时间】：2021-05-10 04:11:08
【问题描述】：

输入数据帧的架构

- employeeKey (int)  
- employeeTypeId (string) 
- loginDate (string)
- employeeDetailsJson (string)

{"Grade":"100","ValidTill":"2021-12-01","Supervisor":"Alex","Vendor":"technicia","HourlyRate":29}

对于 Perm 员工，有些属性可用，有些则不可用。签约员工也是如此。

因此希望找到一种有效的方法来仅基于选定的列构建数据框，而不是转换所有列并选择我需要的列。

另外请告知这是基于键从 json 字符串中提取值的最佳方法。由于字符串中的属性是动态的，我无法基于它构建 StructSchema。所以使用好旧的get_json_object。

（spark 2.45，将来会使用 spark 3）

  val dfSelectColumns=List("Employee-Key", "Employee-Type","Login-Date","cont.Vendor-Name","cont.Hourly-Rate" )

//val dfSelectColumns=List("Employee-Key", "Employee-Type","Login-Date","perm.Level","perm-Validity","perm.Supervisor" )

 val resultDF = inputDF.get
        .withColumn("Employee-Key", col("employeeKey"))
        .withColumn("Employee-Type", when(col("employeeTypeId") === 1, "Permanent")
          .when(col("employeeTypeId") === 2, "Contractor")
          .otherwise("unknown"))  
        .withColumn("Login-Date", to_utc_timestamp(to_timestamp(col("loginDate"), "yyyy-MM-dd'T'HH:mm:ss"), ""America/Chicago""))
        .withColumn("perm.Level", get_json_object(col("employeeDetailsJson"), "$.Grade"))
        .withColumn("perm.Validity", get_json_object(col("employeeDetailsJson"), "$.ValidTill"))
        .withColumn("perm.SuperVisor", get_json_object(col("employeeDetailsJson"), "$.Supervisor"))
        .withColumn("cont.Vendor-Name", get_json_object(col("employeeDetailsJson"), "$.Vendor"))
        .withColumn("cont.Hourly-Rate", get_json_object(col("employeeDetailsJson"), "$.HourlyRate"))
        .select(dfSelectColumns.head, dfSelectColumns.tail: _*)

【问题讨论】：

哦，我不知道。你能告诉我如何接受答案吗？对不起：）

标签： scala apache-spark apache-spark-sql

【解决方案1】：

我看到您有 2 个架构，一个用于永久，另一个用于承包商。您可以有 2 个架构。

import org.apache.spark.sql.types._
import org.apache.spark.sql.functions._

val schemaBase = new StructType().add("Employee-Key", IntegerType).add("Employee-Type", StringType).add("Login-Date", DateType)
val schemaPerm = schemaBase.add("Level", IntegerType).add("Validity", StringType)// Permanent attributes
val schemaCont = schemaBase.add("Vendor", StringType).add("HourlyRate", DoubleType)  // Contractor attributes

然后您可以使用 2 种模式将数据加载到数据框中。
对于永久雇员：

val jsonPermDf = Seq( // Construct sample dataframe
  (2, """{"Employee-Key":2, "Employee-Type":"Permanent", "Login-Date":"2021-11-01", "Level":3, "Validity":"ok"}""")
  , (3, """{"Employee-Key":3, "Employee-Type":"Permanent", "Login-Date":"2020-10-01", "Level":2, "Validity":"ok-yes"}""")
).toDF("key", "raw_json")

val permDf = jsonPermDf.withColumn("data", from_json(col("raw_json"),schemaPerm)).select($"data.*")
permDf.show()

对于承包商：

val jsonContDf = Seq(  // Construct sample dataframe
  (1, """{"Employee-Key":1, "Employee-Type":"Contractor", "Login-Date":"2021-12-01", "Vendor":"technicia", "HourlyRate":29}""")
  , (4, """{"Employee-Key":4, "Employee-Type":"Contractor", "Login-Date":"2019-09-01", "Vendor":"Minis", "HourlyRate":35}""")
).toDF("key", "raw_json")

val contDf = jsonContDf.withColumn("data", from_json(col("raw_json"),schemaCont)).select($"data.*")
contDf.show()

这是永久的结果数据框：

+------------+-------------+----------+-----+--------+
|Employee-Key|Employee-Type|Login-Date|Level|Validity|
+------------+-------------+----------+-----+--------+
|           2|    Permanent|2021-11-01|    3|      ok|
|           3|    Permanent|2020-10-01|    2|  ok-yes|
+------------+-------------+----------+-----+--------+

这是承包商的结果数据框：

+------------+-------------+----------+---------+----------+
|Employee-Key|Employee-Type|Login-Date|   Vendor|HourlyRate|
+------------+-------------+----------+---------+----------+
|           1|   Contractor|2021-12-01|technicia|      29.0|
|           4|   Contractor|2019-09-01|    Minis|      35.0|
+------------+-------------+----------+---------+----------+

【讨论】：

很多时候我需要处理这两种模式。它既不是也不是。

【解决方案2】：

如果 employeeDetailsJson 中 JSON 的 schema 不稳定，您仍然可以使用带有 schema map<string,string> 的 from_json 函数将其解析为 Map(String, String) 类型。然后你可以explode map 列和pivot 来获取键作为列。

例子：

val df1 = df.withColumn(
    "employeeDetails",
    from_json(col("employeeDetailsJson"), "map<string,string>")
  ).select(
    col("employeeKey"),
    col("employeeTypeId"),
    col("loginDate"),
    explode("employeeDetails")
  ).groupBy("employeeKey", "employeeTypeId", "loginDate")
  .pivot("key")
  .agg(first("value"))

df1.show()
//+-----------+--------------+---------------------+-----+----------+----------+----------+---------+
//|employeeKey|employeeTypeId|loginDate            |Grade|HourlyRate|Supervisor|ValidTill |Vendor   |
//+-----------+--------------+---------------------+-----+----------+----------+----------+---------+
//|1          |1             |2021-02-05'T'21:28:06|100  |29        |Alex      |2021-12-01|technicia|
//+-----------+--------------+---------------------+-----+----------+----------+----------+---------+

【讨论】：