如何在PySpark中分解嵌套数据框并将其进一步存储到配置单元 [英] how to explode Nested data frame in PySpark and further store it to hive

问题描述

我面临一个特定的问题，其中我的JSON结构如下:-

I am facing a certain issue in which i have my JSON structure as follows :-

{A:value,
B:value,
C:Array<Struct<A1:value,B1:value, C1:Array<struct<A2:value,B2:value>>>>
}

我希望将其分解为以下形式:-

i would want it to be exploded in the following form:-

{
A:value,
B:value,
A1:value,
B1:value,
A2:value,
B2:value
}

我为此使用了pyspark数据帧，找不到正确爆炸的方法.任何帮助表示赞赏.

I am using pyspark dataframes for this and couldn't find a way to explode properly. any help is appreciated.

推荐答案

让我们从一个示例数据帧开始，该数据帧具有与您指定的模式相同的模式:

Let's start with a sample dataframe, with the same schema as the one you specified:

import pyspark.sql.functions as psf
df = sc.parallelize([["a", "b", "c.a1", "c.b1", "c.c1.a2", "c.c1.a3"]]).toDF(["A", "B", "A1", "B1", "A2", "B2"])\
    .select("A", "B", psf.array(psf.struct("A1", "B1", psf.array(psf.struct("A2", "B2")).alias("C1"))).alias("C"))
df.printSchema()

    root
     |-- A: string (nullable = true)
     |-- B: string (nullable = true)
     |-- C: array (nullable = false)
     |    |-- element: struct (containsNull = false)
     |    |    |-- A1: string (nullable = true)
     |    |    |-- B1: string (nullable = true)
     |    |    |-- C1: array (nullable = false)
     |    |    |    |-- element: struct (containsNull = false)
     |    |    |    |    |-- A2: string (nullable = true)
     |    |    |    |    |-- B2: string (nullable = true)

选择列时，可以使用*分解StructType内联元素(例如，如果C1是StrucType，则是select("C1.*")).您的情况有些复杂，因为这些StructType嵌套在ArrayType中.但是，在 Spark2 中，您可以访问ArrayType中包含的StructType的嵌套元素，输出将是这些元素的ArrayType:

You can explode the elements of a StructType inline using * when selecting the column (ie. select("C1.*") if C1 were a StrucType) . Your case is a bit more complex since these StructTypes are nested in an ArrayType. Nonetheless, in Spark2, you can access the nested elements of StructType contained in an ArrayType, the output will be an ArrayType of these elements:

df.select("C.A1").show()

    +------+
    |    A1|
    +------+
    |[c.a1]|
    +------+

您可以使用以下功能自动执行此过程:

You can automatize this process by using the following functions:

获取列的嵌套列:

def get_subcols(df, col):
    if col in df.columns:
        subschema = [s["type"]["elementType"]["fields"] for s in df.schema.jsonValue()["fields"] if s["name"] == col][0]
        return  [s["name"] for s in subschema]
    else:
        return None

将数据框展平为一个级别:

to flatten a dataframe for one level:

import re 
def flatten_df(df):
    non_nested_cols = [c[0] for c in df.dtypes if not re.match("array<struct|struct", c[1])]
    nested_cols = [c[0] for c in df.dtypes if re.match("array<struct|struct", c[1])]
    return df.select(non_nested_cols + [psf.col(c1 + "." + c2) for c1 in nested_cols for c2 in get_subcols(df, c1)])

由于数据帧需要展平两次，因此您将不得不在某些时候使用explode，因为您将得到ArrayType或StructType的ArrayType:

Since your dataframe needs to be flattened twice, you will have to use explode at some point, since you'll get an ArrayType of an ArrayType of a StructType:

df1 = flatten_df(df)
df1.printSchema()

    root
     |-- A: string (nullable = true)
     |-- B: string (nullable = true)
     |-- A1: array (nullable = false)
     |    |-- element: string (containsNull = false)
     |-- B1: array (nullable = false)
     |    |-- element: string (containsNull = false)
     |-- C1: array (nullable = false)
     |    |-- element: array (containsNull = false)
     |    |    |-- element: struct (containsNull = false)
     |    |    |    |-- A2: string (nullable = true)
     |    |    |    |-- B2: string (nullable = true)

对于 Spark1 ，您每次必须使用explode:

For Spark1, you have to use explode everytime:

df.select("A", "B", psf.explode("C").alias("C"))\
    .select("A", "B", "C.*")\
    .select("A", "B", "A1", "B1", psf.explode("C1").alias("C1"))\
    .select("A", "B", "A1", "B1", "C1.*")\
    .show()

    +---+---+----+----+-------+-------+
    |  A|  B|  A1|  B1|     A2|     B2|
    +---+---+----+----+-------+-------+
    |  a|  b|c.a1|c.b1|c.c1.a2|c.c1.a3|
    +---+---+----+----+-------+-------+

请注意，explode会创建与数组中的行一样多的行(此处每个数组中只有一个元素).然后，您可以使用A, B作为键，group将数据框返回.

Note that explode creates as many lines as there are in your array (here there is only one element in every array). Then you can group you dataframe back using A, B as keys for instance.

这篇关于如何在PySpark中分解嵌套数据框并将其进一步存储到配置单元的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！