在 Pyspark 中为 json 数据定义模式

Question

我有一个 json 如下存储在 json 文件中的 amazon s3 中。

{
    "owner": "mindey",
    "ownerId": "4e751a",
    "batch": "test1",
    "mode": "testserve",
    "subType": "self",
    "versionId": "v89.0563",
    "createdDate": "2021-12-10T00:41:02.652Z",
    "modifiedDate": "2021-12-10T00:41:02.652Z",
    "createdBy": "xyz@nike.com",
    "modifiedBy": "xyz@nike.com",
    "numChanges": 10,
    "progressFlag": true,
    "auditing": {
        "hack": "TBD",
        "facetCd": [
            "652439",
            "A97341"
        ],
        "modelColors": [
            "083y32-058",
            "978t22-456",
            "A097ee-009"
        ],
        "consumersGroup": [
            "25a9874-098a-08nh-4424-90838e"
        ],
        "cnsmrTargets": [
            {
                "id": "7jsfg8734-038h-9825-9784-37768ebe6ba1",
                "type": "self"
            }
        ],
        "modelGroups": [
            {
                "seedIds": [
                    "7kjshf99-754a-gf78-a015",
                    "6d874mq3-23fa-4e76-baad"
                ],
                "exact": true,
                "position": 5
            }
        ],
        "txGroups": [
            {
                "txIds": [
                    "7bw97862-aca7-9873",
                    "5ddfb803-2f11-8763"
                ]
            }
        ],
        "shops": [
            "ue749nt2-acc6-2452",
            "ue749nt2-982y-4286"
        ]
    }
}

现在，我阅读了文件并希望在 pyspark 中定义一个模式以将数据附加到表中。因此，我将我的模式定义如下，并通过 pyspark 中的临时数据框读取它，如下所示。代码没有出错，但是我希望得到一个不同的结果，下面也提到了

sch = StructType([StructField("owner",StringType()), 
        StructField("ownerId",StringType()), 
        StructField("batch",StringType()), 
        StructField("mode",StringType()), 
        StructField("subType",StringType()), 
        StructField("versionId",StringType()), 
        StructField("createdDate",StringType()), 
        StructField("modifiedDate",StringType()), 
        StructField("createdBy",StringType()), 
        StructField("modifiedBy",StringType()), 
        StructField("numChanges",StringType()), 
        StructField("progressFlag",StringType()), 
        StructField("auditing",StringType())])


df_read = spark.read.json(src,schema=sch)
df_read.createOrReplaceTempView('df')
spark.sql("select * from df").show(10,False)

+------+-------+-----+----+-------+---------+------------------------+------------------------+------------+------------+----------+------------+-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
|owner |ownerId|batch|mode|subType|versionId|createdDate             |modifiedDate            |createdBy   |modifiedBy  |numChanges|progressFlag|auditing                                                                                                                                                                                                                                                                                                                                                                                                                                                               |
+------+-------+-----+----+-------+---------+------------------------+------------------------+------------+------------+----------+------------+-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
|mindey|4e751a |test1|null|self   |v89.0563 |2021-12-10T00:41:02.652Z|2021-12-10T00:41:02.652Z|xyz@nike.com|xyz@nike.com|10        |true        |{"hack":"TBD","facetCd":["652439","A97341"],"modelColors":["083y32-058","978t22-456","A097ee-009"],"consumersGroup":["25a9874-098a-08nh-4424-90838e"],"cnsmrTargets":[{"id":"7jsfg8734-038h-9825-9784-37768ebe6ba1","type":"self"}],"modelGroups":[{"seedIds":["7kjshf99-754a-gf78-a015","6d874mq3-23fa-4e76-baad"],"exact":true,"position":5}],"txGroups":[{"txIds":["7bw97862-aca7-9873","5ddfb803-2f11-8763"]}],"shops":["ue749nt2-acc6-2452","ue749nt2-982y-4286"]}|
+------+-------+-----+----+-------+---------+------------------------+------------------------+------------+------------+----------+------------+-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+

我希望输出如下所示，其中嵌套的 json 被适当地拆分到不同的列 -

+------+-------+-----+----+-------+---------+------------------------+------------------------+------------+------------+----------+------------+-------------+--------------------+--------------------------------------------+--------------------------------------------+--------------------------------------------+--------------------------------------------+--------------------------------------------------------+--------------------------------------------------------+-------------------------------------------------+
|owner |ownerId|batch|mode|subType|versionId|createdDate             |modifiedDate            |createdBy   |modifiedBy  |numChanges|progressFlag|auditing_hack|auditing_facetCd    |auditing_modelColors                        |auditing_consumersGroup                     |auditing_cnsmrTargets_id                    |auditing_cnsmrTargets_type                  |auditing_modelGroups_seedIds                            |auditing_txGroups_txIds                                 |auditing_txGroups_shops                          |
+------+-------+-----+----+-------+---------+------------------------+------------------------+------------+------------+----------+------------+-------------+--------------------+--------------------------------------------+--------------------------------------------+--------------------------------------------+--------------------------------------------+--------------------------------------------------------+--------------------------------------------------------+-------------------------------------------------+
|mindey|4e751a |test1|null|self   |v89.0563 |2021-12-10T00:41:02.652Z|2021-12-10T00:41:02.652Z|xyz@nike.com|xyz@nike.com|10        |true        |TBD          |["652439","A97341"] |["083y32-058","978t22-456","A097ee-009"]    |25a9874-098a-08nh-4424-90838e               |7jsfg8734-038h-9825-9784-37768ebe6ba1       |self                                        |["7kjshf99-754a-gf78-a015","6d874mq3-23fa-4e76-baad"]   |["7bw97862-aca7-9873","5ddfb803-2f11-8763"]             |["ue749nt2-acc6-2452","ue749nt2-982y-4286"]}     |
+------+-------+-----+----+-------+---------+------------------------+------------------------+------------+------------+----------+------------+-------------+--------------------+--------------------------------------------+--------------------------------------------+--------------------------------------------+--------------------------------------------+--------------------------------------------------------+--------------------------------------------------------+-------------------------------------------------+

有人可以帮我解决这个问题吗？提前致谢！！

Answer 1

架构对象应如下所示：

sch = StructType([
  StructField("owner",StringType()), 
  StructField("ownerId",StringType()), 
  StructField("batch",StringType()), 
  StructField("mode",StringType()), 
  StructField("subType",StringType()), 
  StructField("versionId",StringType()), 
  StructField("createdDate",StringType()), 
  StructField("modifiedDate",StringType()), 
  StructField("createdBy",StringType()), 
  StructField("modifiedBy",StringType()), 
  StructField("numChanges",StringType()), 
  StructField("progressFlag",StringType()), 
  StructField("auditing",StructType([
    StructField('hack', StringType()),
    StructField('facetCd', ArrayType(StringType())),
    StructField('modelColors', ArrayType(StringType())),
    StructField('consumersGroup', ArrayType(StringType())),
    StructField('cnsmrTargets', ArrayType(StringType())),
    StructField('modelGroups', ArrayType(StructType([
      StructField('seedIds', ArrayType(StringType())),
      StructField('exact', StringType()),
      StructField('position', IntegerType()),
    ]))),
    StructField('facetCd', ArrayType(StructType([
      StructField('txIds', ArrayType(StringType())),
    ]))),
    StructField('shops', ArrayType(StringType())),
  ]))
 ]
)

应用架构后，您将获得一个用于审计列的嵌套结构。然后结合使用 select 语句和别名可以获得所需的列名。