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

debugcn 发表于 Dev

酷鹅

输入数据框的架构

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

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

对于彼尔姆员工，某些属性可用，而有些则不可用。与签约员工相同。

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

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

（火花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: _*)

空中黄油

我看到您有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|
+------------+-------------+----------+---------+----------+

本文收集自互联网，转载请注明来源。

如有侵权，请联系[email protected] 删除。