如何从PySpark数据框中提取嵌套JSON的列值？

Question

我正在处理一个PySpark的数据框（es_query），里面有一些嵌套的JSON列（r_json、brd_json、vs_json）。我需要帮助来提取这些列的数据，并把它们存储到另一个数据框（e_result）中，作为两列，分别是URL和产品编号，每一行都是一个单独的记录。

最后，我们需要把所有的值放到一个数据框里。

上面提到的各个列的示例数据如下：

r_json:
results:
0: {"col1": "Yes", "name": "", "col2":  1, "col3": "76,67 €", "col4": "5,75 €", "productNumber": "B0e28213", "url": "https://www.am"}
1: {"col1": "Yes", "name": "", "col2":  1, "col3": "76,67 €", "col4": "5,75 €", "productNumber": "019883", "url": "https://www.am"}

brd_json:
array:
0: {"col1": "Yes", "col2": "https://m.media-a", "col3": null, "col4": "Yes", , "col5": "No", "col6": false, "productNumber": "11873628", "rating": "4.1", "url": "https://www.amazon"}
1: {"col1": "Yes", "col2": "https://m.media-a", "col3": null, "col4": "Yes", , "col5": "No", "col6": false, "productNumber": "001838", "rating": "4.1", "url": "https://www.amazon"}

vs_json:
array:
0: 0: {"col1": "Yes", "col2": "https://m.media-a", "col3": null, "col4": "Yes", , "col5": "No", "col6": false, "productNumber": "1212", "rating": "4.1", "url": "https://www.amazon"}
1: 0: {"col1": "Yes", "col2": "https://m.media-a", "col3": null, "col4": "Yes", , "col5": "No", "col6": false, "productNumber": "2321", "rating": "4.1", "url": "https://www.amazon"}

我能得到一些帮助来写这个脚本吗？下面是我尝试过的：

from pyspark.sql import SparkSession
from pyspark.sql.functions import lit

result_urls = results.select("url").withColumn("source", lit("result"))
brsd_json_url = brsd_json.select("url").withColumn("source", lit("brand"))
vis_json_url = brsd_json.select("url").withColumn("source", lit("video"))

combined_urls = result_urls.union(brsd_json_url).union(vis_json_url)

esearch_result = combined_urls.groupBy("source").agg({"url": "concat_ws"}).select("concat_ws(url) as prod_page_url_address")

我尝试了下面的代码，这能帮助提取所有列作为独立的PySpark数据框，但现在的问题是如何把它们合并成一个数据框，同时考虑到所有的值。

column_names = ["url", "productNumber", "col2", "col3", "col4", "col5", 
                "col6", "col7"]

# Define a function to create the expressions
def create_expr(column_name, json_column):
    try:
        expr = esearch_request_query.selectExpr(f"EXPLODE({json_column}.{column_name}) as {column_name}")
    except:
        expr = None
    return expr

# Iterate over the column names
for column_name in column_names:
    results_expr = create_expr(column_name, "result_json.re")
    brsd_json_url  = create_expr(column_name, "brsd_json")
    vis_json_url  = create_expr(column_name, "vis_json")

    globals()[f"results_{column_name}"] = results_expr
    globals()[f"brsd_json_{column_name}"] = brsd_json_url
    globals()[f"vis_json_{column_name}"] = vis_json_url

# case where "col2" does not exist in sponsored_video_json
try:
    results_brandName = esearch_request_query.selectExpr("EXPLODE(result_json.re.col2) as col2_name")
except:
    results_brandName = None

try:
    sponnd_col2 = esearch_request_query.selectExpr("EXPLODE(brsd_json.spon) as col2_name")
except:
    spond_col2 = None

spovideo_col2 = None  # Handle this case separately, as it does not exist in vis_json

数据处理数据提取 json 数据框数据合并列操作 pyspark 嵌套数据

如何从PySpark数据框中提取嵌套JSON的列值？

1 个回答

撰写回答