下面是我的源架構。
root
|-- header: struct (nullable = true)
| |-- timestamp: long (nullable = true)
| |-- id: string (nullable = true)
| |-- honame: string (nullable = true)
|-- device: struct (nullable = true)
| |-- srcId: string (nullable = true)
| |-- srctype.: string (nullable = true)
|-- ATTRIBUTES: map (nullable = true)
| |-- key: string
| |-- value: string (valueContainsNull = true)
|-- event_date: date (nullable = true)
|-- event_datetime: string (nullable = true)
我想分解 ATTRIBUTES 映射型別列并選擇所有以 _id 結尾的列。
我使用下面的代碼。
val exploded = batch_df.select($"event_date", explode($"ATTRIBUTES")).show()
我得到以下示例輸出。
--- ---------- -------------------- --------------------
|date | key| value|
---------- -------------------- --------------------
|2021-05-18|SYST_id | 85|
|2021-05-18|RECVR_id | 1|
|2021-05-18|Account_Id| | 12345|
|2021-05-18|Vb_id | 845|
|2021-05-18|SYS_INFO_id | 640|
|2021-05-18|mem_id | 456|
------------------------------------------------------
但是，我需要的輸出如下。
--- ------- -------------- ----------- ------------ ------- ------------- -------
|date | SYST_id | RECVR_id | Account_Id | Vb_id | SYS_INFO_id| mem_id|
---- ------ -------------- ----------- ------------ ------- ------------- -------
|2021-05-18| 85 | 1 | 12345 | 845 | 640 | 456 |
----------- -------------- ----------- ------------ ------- ------------- -------
有人可以幫忙嗎？
uj5u.com熱心網友回復：
你的方法有效。您只需要在 之后添加一個pivot操作explode：
import org.apache.spark.sql.functions._
exploded.groupBy("date").pivot("key").agg(first("value")).show()
我假設的組合date和key唯一性，因此它是安全采取first的聚集（只）值。如果組合不是唯一的，則可以collect_list用作聚合函式。
編輯：
要添加scrIdand srctype，只需將這些列添加到select陳述句中：
val exploded = batch_df.select($"event_date", $"device.srcId", $"device.srctype", explode($"ATTRIBUTES"))
要減少pivot操作后的列數，請key在聚合前對列應用過濾器：
val relevant_cols = Array("Account_Id", "Vb_id", "RECVR_id", "mem_id") // the four additional columns
exploded.filter($"key".isin(relevant_cols:_*).or($"key".endsWith(lit("_split"))))
.groupBy("date").pivot("key").agg(first("value")).show()
轉載請註明出處，本文鏈接：https://www.uj5u.com/ruanti/311451.html