Pyspark: Split plusieurs colonnes de tableau en lignes

J'ai un dataframe qui a une ligne, et plusieurs colonnes. Certaines colonnes sont des valeurs individuelles, et d'autres sont des listes. Toutes les colonnes de la liste sont de la même longueur. Je veux diviser chaque colonne de la liste dans une ligne distincte, tout en conservant la non-colonne de la liste tel qu'il est.

Échantillon DF:

from pyspark import Row
from pyspark.sql import SQLContext
from pyspark.sql.functions import explode

sqlc = SQLContext(sc)

df = sqlc.createDataFrame([Row(a=1, b=[1,2,3],c=[7,8,9], d='foo')])
# +---+---------+---------+---+
# |  a|        b|        c|  d|
# +---+---------+---------+---+
# |  1|[1, 2, 3]|[7, 8, 9]|foo|
# +---+---------+---------+---+

Ce que je veux:

+---+---+----+------+
|  a|  b|  c |    d |
+---+---+----+------+
|  1|  1|  7 |  foo |
|  1|  2|  8 |  foo |
|  1|  3|  9 |  foo |
+---+---+----+------+

Si je n'avais qu'une colonne de la liste, ce serait facile, il vous suffit de faire un explode:

df_exploded = df.withColumn('b', explode('b'))
# >>> df_exploded.show()
# +---+---+---------+---+
# |  a|  b|        c|  d|
# +---+---+---------+---+
# |  1|  1|[7, 8, 9]|foo|
# |  1|  2|[7, 8, 9]|foo|
# |  1|  3|[7, 8, 9]|foo|
# +---+---+---------+---+

Cependant, si j'essaie aussi de explode la c colonne, je me retrouve avec un dataframe avec une longueur au carré de ce que je veux:

df_exploded_again = df_exploded.withColumn('c', explode('c'))
# >>> df_exploded_again.show()
# +---+---+---+---+
# |  a|  b|  c|  d|
# +---+---+---+---+
# |  1|  1|  7|foo|
# |  1|  1|  8|foo|
# |  1|  1|  9|foo|
# |  1|  2|  7|foo|
# |  1|  2|  8|foo|
# |  1|  2|  9|foo|
# |  1|  3|  7|foo|
# |  1|  3|  8|foo|
# |  1|  3|  9|foo|
# +---+---+---+---+

Est ce que je veux - pour chaque colonne, prendre la n-ième élément du tableau dans la colonne et l'ajouter à une nouvelle ligne. J'ai essayé de cartographie d'un exploser à travers toutes les colonnes dans le dataframe, mais cela ne semble pas fonctionner soit:

df_split = df.rdd.map(lambda col: df.withColumn(col, explode(col))).toDF()
InformationsquelleAutor Steve | 2016-12-07
  1. 50

    Spark >= 2.4

    Vous pouvez remplacer zip_ udf avec arrays_zip fonction

    from pyspark.sql.functions import arrays_zip, col

(df
    .withColumn("tmp", arrays_zip("b", "c"))
    .withColumn("tmp", explode("tmp"))
    .select("a", col("tmp.b"), col("tmp.c"), "d"))

    Spark < 2.4

    Avec DataFrames et de l'UDF:

    from pyspark.sql.types import ArrayType, StructType, StructField, IntegerType
from pyspark.sql.functions import col, udf, explode

zip_ = udf(
  lambda x, y: list(zip(x, y)),
  ArrayType(StructType([
      # Adjust types to reflect data types
      StructField("first", IntegerType()),
      StructField("second", IntegerType())
  ]))
)

(df
    .withColumn("tmp", zip_("b", "c"))
    # UDF output cannot be directly passed to explode
    .withColumn("tmp", explode("tmp"))
    .select("a", col("tmp.first").alias("b"), col("tmp.second").alias("c"), "d"))

    Avec RDDs:

    (df
    .rdd
    .flatMap(lambda row: [(row.a, b, c, row.d) for b, c in zip(row.b, row.c)])
    .toDF(["a", "b", "c", "d"]))

    Les deux solutions sont inefficaces en raison de Python communication frais généraux. Si la taille des données est fixe, vous pouvez faire quelque chose comme ceci:

    from functools import reduce
from pyspark.sql import DataFrame

# Length of array
n = 3

# For legacy Python you'll need a separate function
# in place of method accessor 
reduce(
    DataFrame.unionAll, 
    (df.select("a", col("b").getItem(i), col("c").getItem(i), "d")
        for i in range(n))
).toDF("a", "b", "c", "d")

    ou encore:

    from pyspark.sql.functions import array, struct

# SQL level zip of arrays of known size
# followed by explode
tmp = explode(array(*[
    struct(col("b").getItem(i).alias("b"), col("c").getItem(i).alias("c"))
    for i in range(n)
]))

(df
    .withColumn("tmp", tmp)
    .select("a", col("tmp").getItem("b"), col("tmp").getItem("c"), "d"))

    Cela devrait être beaucoup plus rapide par rapport à l'UDF ou de RDD. Généralisée à l'appui d'un nombre quelconque de colonnes:

    # This uses keyword only arguments
# If you use legacy Python you'll have to change signature
# Body of the function can stay the same
def zip_and_explode(*colnames, n):
    return explode(array(*[
        struct(*[col(c).getItem(i).alias(c) for c in colnames])
        for i in range(n)
    ]))

df.withColumn("tmp", zip_and_explode("b", "c", n=3))
    InformationsquelleAutor user6910411
  2. 8

    Vous auriez besoin d'utiliser flatMap, pas map que vous voulez faire de plusieurs lignes de sortie de chaque ligne d'entrée.

    from pyspark.sql import Row
def dualExplode(r):
    rowDict = r.asDict()
    bList = rowDict.pop('b')
    cList = rowDict.pop('c')
    for b,c in zip(bList, cList):
        newDict = dict(rowDict)
        newDict['b'] = b
        newDict['c'] = c
        yield Row(**newDict)

df_split = sqlContext.createDataFrame(df.rdd.flatMap(dualExplode))
    • si la première df a 3 valeurs et deuxième df a 2 valeurs, notre zip se trouve être le retour de deux paires au lieu de 3. Pourriez-vous des conseils sur elle.
    • Zip paires ensemble le premier élément d'un obj avec le 1er élément d'un autre objet, de la 2e à la 2ème, etc jusqu'à ce que l'un des objets est à court d'éléments. Dans votre cas, après 2 valeurs. Dit d'une autre manière, il sera paire des éléments jusqu'à ce qu'il n'y a pas plus d'éléments pour paire. Pour donner à toutes les suggestions, j'aurais besoin de savoir comment vous voulez que votre programme pour traiter avec les nations unies, jumelé élément (par exemple, voulez-vous une valeur null à partir de la 2e?). Aussi, il est à seulement 1 df dans cet exemple. Si votre question est-ce différent de celui-ci, il est probablement préférable de simplement poser une autre question
    • Merci @David pour votre réponse. J'ai pensé à elle. À l'aide de Izip aidé à résoudre ce problème. Mais encore, je vous remercie de votre réponse mate.
    InformationsquelleAutor David