Pyspark 爆炸功能未按预期工作

Question

我正在将一个配置单元表读入 Pyspark 数据帧。 hive 表有一个定义为字符串的有效负载。样例记录，

{
  "journal": {
    "batchName": "Test",
    "lines": [
      {
        "lineNumber": 1,
        "lineDescription": "Something",
        "companyCode": "100",
        "debit": 0,
        "credit": 27,
        "accountedDebit": 0,
        "accountedCredit": 27
      },
      {
        "lineNumber": 2,
        "lineDescription": "Somethingg",
        "companyCode": "100",
        "debit": 27,
        "credit": 0,
        "accountedDebit": 27,
        "accountedCredit": 0
      }
    ]
  }
}

这是存储为字符串。我能够使用 get_json_object() 提取 batchName 但无法将行扩展为 2 行。我试图将线条转换为数组，然后使用explode，但它仍然不起作用。示例代码如下。

main_data3 = main_data2.withColumn("lines_line_number2", F.regexp_replace(F.regexp_replace(F.col("lines_line_number"), '\[', ''),'\]',''))
main_data4 = main_data3.withColumn("lines_line_number3", F.array("lines_line_number2"))
main_data5 = main_data4.withColumn("lines_line_number4", F.explode("lines_line_number3")
... )

我想使用 spark sql 从这个数据集中扩展和提取所有列。我无法定义自定义架构。

Answer 1

您可以使用outer_explode，然后使用简单的“*”将结构中的所有字段作为单独的列。

我根据您提供的字符串创建了一个数据框df。

    from pyspark.sql.functions import explode_outer

    df_json = spark.read.json(df.rdd.map(lambda r: r.json))
    # Explode the lines array, which will result in struct.
    exploded_df =df_json.select(df_json.journal.batchName.alias('batchname'),explode_outer(df_json.journal.lines).alias('exploded_lines'))
    
    # Select all fields from the exploded struct.
    display(new_df_2.select('batchname', 'exploded_lines.*'))