列印資料框中不同的列名

我有這個資料框

 ----- ------- ----------- ------------------- ----- 
|empID|Zipcode|ZipCodeType|City               |State|
 ----- ------- ----------- ------------------- ----- 
|1000 |704    |STANDARD   |PARC PARQUE        |PR   |
|1000 |704    |STANDARD   |PASEO COSTA DEL SUR|PR   |
|1001 |709    |STANDARD   |BDA SAN LUIS       |PR   |
|1001 |76166  |UNIQUE     |CINGULAR WIRELESS  |TX   |
|1002 |76177  |STANDARD   |FORT WORTH         |TX   |
|1002 |76177  |STANDARD   |FT WORTH           |TX   |
|1003 |704    |STANDARD   |URB EUGENE RICE    |PR   |
|1003 |85209  |STANDARD   |MESA               |AZ   |
|1004 |85210  |STANDARD   |MESA               |AZ   |
|1004 |32046  |STANDARD   |HILLIARD           |FL   |
 ----- ------- ----------- ------------------- ----- 

對于每個 empID 需要列印其值不同的列名。

 ----- --------------------------------- 
|empID|nonMatchingColumnNames           |
 ----- --------------------------------- 
|1002 |City                             |
|1000 |City                             |
|1001 |State, City, ZipCodeType, Zipcode|
|1003 |State, City, Zipcode             |
|1004 |State, City, Zipcode             |
 ----- --------------------------------- 

我采取的策略是，構建一個結構并收集所有值。檢查每個集合的計數是否> 1，然后列印列名稱。這是我的代碼

val schema = new StructType()
  .add("empID", IntegerType, true)
  .add("Zipcode", StringType, true)
  .add("ZipCodeType", StringType, true)
  .add("City", StringType, true)
  .add("State", StringType, true)
    
val idColumn = "empID"
    
val dfJSON = dfFromText.withColumn("jsonData",from_json(col("value"),schema))
  .select("jsonData.*")
    
dfJSON.printSchema()
dfJSON.show(false)
    
val aggMap = dfJSON.columns
  .filterNot(x => x == idColumn)
  .map(colName => (collect_set(colName).alias(s"${colName}_asList"), s"${colName}_asList"))
   
aggMap.foreach(println)
    
val aggMapColumns = aggMap.map(x => x._1)
    
val columnsAsList = dfJSON.groupBy(col(idColumn)).agg(aggMapColumns.head, aggMapColumns.tail : _ *)
    
columnsAsList.show(false)
    
val combinedDF = columnsAsList.select(col(idColumn), struct(
  aggMap.map(x => col(x._2)) : _ * ).alias("combined_struct")
)
    
combinedDF.printSchema()
combinedDF.show(false)
    
val columnsToCompare = dfJSON.columns.filterNot(x => x == idColumn).zipWithIndex.map({ case (x,y) => (y,x)})
    
val output = combinedDF.rdd.map({row => {
  val empNo = row.getAs[Int](0)
  val conbinedStruct: Row = row.getAs[AnyRef]("combined_struct").asInstanceOf[Row]
    
  val nonMatchingColumns = columnsToCompare.foldLeft(List[String]())((acc, item) => {
    val counts = conbinedStruct.getAs[Seq[String]](item._1).length
    if (counts == 1) acc else item._2 :: acc
  })
    
  (empNo, nonMatchingColumns.mkString(", "))
}}).toDF(idColumn, "nonMatchingColumnNames")
    
output.show(false)

它在我的本地機器上作業得非常好，當我將它移植到 spark-shell（它是一個臨時查詢）時，當我嘗試將資料幀轉換為 RDD 并遍歷結構中的每個專案時，我遇到了空指標例外。

uj5u.com熱心網友回復：

您只能使用 spark 的內置函式來獲取包含值不唯一的列串列的字串：

• 用于countDistinct確定特定列中是否有特定列中的多個值empID
• 如果 count distinct 大于 2，則保存列的名稱，使用 when
• 迭代列并將此迭代保存到陣列中 array
• 使用此陣列構建一個字串 concat_ws

完整代碼如下：

import org.apache.spark.sql.functions.{array, concat_ws, countDistinct, lit, when}

val output = dfJSON.groupBy("empID").agg(
  concat_ws(
    ", ",
    array(dfJSON.columns.filter(_ != "empID").map(c => when(countDistinct(c) > 1, lit(c))): _*)
  ).as("nonMatchingColumnNames")
)

使用您的輸入資料框，您將獲得以下輸出：

 ----- --------------------------------- 
|empID|nonMatchingColumnNames           |
 ----- --------------------------------- 
|1002 |City                             |
|1000 |City                             |
|1001 |Zipcode, ZipCodeType, City, State|
|1003 |Zipcode, City, State             |
|1004 |Zipcode, City, State             |
 ----- --------------------------------- 

標籤：斯卡拉 阿帕奇火花 apache-spark-sql

上一篇：為什么此欄位驗證型別不匹配？

下一篇：如何在Scala中初始化空的`util.Collection[String]`？