pyspark：將dict陣列轉換為新列

我正在努力轉換我的 pyspark 資料框，如下所示：

df = spark.createDataFrame([('0018aad4',[300, 450], ['{"v1": "blue"}', '{"v2": "red"}']), ('0018aad5',[300], ['{"v1": "blue"}'])],[ "id","Tlist", 'Tstring'])
df.show(2, False)

 -------- ---------- ------------------------------- 
|id      |Tlist     |Tstring                        |
 -------- ---------- ------------------------------- 
|0018aad4|[300, 450]|[{"v1": "blue"}, {"v2": "red"}]|
|0018aad5|[300]     |[{"v1": "blue"}]               |
 -------- ---------- ------------------------------- 

對此：

df_result = spark.createDataFrame([('0018aad4',[300, 450], 'blue', 'red'), ('0018aad5',[300], 'blue', None)],[ "id","Tlist", 'v1', 'v2'])
df_result.show(2, False)

 -------- ---------- ---- ---- 
|id      |Tlist     |v1  |v2  |
 -------- ---------- ---- ---- 
|0018aad4|[300, 450]|blue|red |
|0018aad5|[300]     |blue|null|
 -------- ---------- ---- ---- 

我試圖旋轉和其他一些東西，但沒有得到上面的結果。

請注意，我沒有列中 dict 的確切數量Tstring

你知道我該怎么做嗎？

uj5u.com熱心網友回復：

使用transform函式，您可以將陣列的每個元素轉換為映射型別。之后，您可以使用aggregate函式獲取一張地圖，將其分解，然后旋轉鍵以獲得所需的輸出：

from pyspark.sql import functions as F

df1 = df.withColumn(
    "Tstring",
    F.transform("Tstring", lambda x: F.from_json(x, "map<string,string>"))
).withColumn(
    "Tstring",
    F.aggregate(
        F.expr("slice(Tstring, 2, size(Tstring))"), 
        F.col("Tstring")[0], 
        lambda acc, x: F.map_concat(acc, x)
    )
).select(
    "id", "Tlist", F.explode("Tstring")
).groupby(
    "id", "Tlist"
).pivot("key").agg(F.first("value"))


df1.show()
# -------- ---------- ---- ---- 
#|id      |Tlist     |v1  |v2  |
# -------- ---------- ---- ---- 
#|0018aad4|[300, 450]|blue|red |
#|0018aad5|[300]     |blue|null|
# -------- ---------- ---- ---- 

我使用的是 Spark 3.1 ，因此資料幀 API 中提供了更高階的函式，transform但您可以使用exprspark <3.1 執行相同的操作。

uj5u.com熱心網友回復：

from pyspark.sql import SparkSession
from pyspark.sql import functions as F

from pyspark.sql.types import *
from datetime import datetime
from pyspark.sql import *
from collections import *
from pyspark.sql.functions import udf,explode
from pyspark.sql.types import StringType
from pyspark.context import SparkContext
from pyspark.sql.session import SparkSession

spark = SparkSession(sc)

df= spark.createDataFrame(
    [
        ('0018aad4', [{"val1":"blue", "val2":"red"}],[300,500]), 
         ('0018aad', [{"val1":"blue", "val2":"null"}],[300])
       
        ],("ID","List","Tlist")
    )

df2 = df.select(df.ID,explode(df.List).alias("Dict"),df.Tlist )
df2.withColumn("Val1", F.col("Dict").getItem("val1")).withColumn("Val2", F.col("Dict").getItem("val2")).show(truncate=False)


 -------- ---------------------------- ---------- ---- ---- 
|ID      |Dict                        |Tlist     |Val1|Val2|
 -------- ---------------------------- ---------- ---- ---- 
|0018aad4|{val2 -> red, val1 -> blue} |[300, 500]|blue|red |
|0018aad |{val2 -> null, val1 -> blue}|[300]     |blue|null|

-------- ---------------- ---------- ---- ----

這就是你要找的。

uj5u.com熱心網友回復：

稍微過度擬合示例（您可能需要對其進行任何泛化調整），您可以Tstring使用它們的索引從列中獲取元素：

partial_results = df.withColumn("v1", df.Tstring[0]).withColumn("v2", df.Tstring[1])

 -------- ---------- -------------- ------------- 
|      id|     Tlist|            v1|           v2|
 -------- ---------- -------------- ------------- 
|0018aad4|[300, 450]|{"v1": "blue"}|{"v2": "red"}|
|0018aad5|     [300]|{"v1": "blue"}|         null|
 -------- ---------- -------------- ------------- 

有了這個，你可以做一些清潔來達到想要的結果

from pyspark.sql.functions import regexp_replace


maximum_color_length = 100
wanted_df = df.withColumn(
    "v1",
    regexp_replace(df.Tstring[0].substr(9, maximum_color_length), r"\"\}", "")
).withColumn(
    "v2",
    regexp_replace(df.Tstring[1].substr(9, maximum_color_length), r"\"\}", "")
).drop(
    "Tstring"
)

 -------- ---------- ---- ---- 
|      id|     Tlist|  v1|  v2|
 -------- ---------- ---- ---- 
|0018aad4|[300, 450]|blue| red|
|0018aad5|     [300]|blue|null|
 -------- ---------- ---- ---- 

標籤：Python 阿帕奇火花 pyspark apache-spark-sql

上一篇：檢查一個詞是否是另一個詞的一部分

下一篇：QStyledItemDelegate/QListView的QAbstractItemDelegate