Pivot de la Chaîne de colonne sur Pyspark Dataframe

J'ai un simple dataframe comme ceci:

rdd = sc.parallelize(
    [
        (0, "A", 223,"201603", "PORT"), 
        (0, "A", 22,"201602", "PORT"), 
        (0, "A", 422,"201601", "DOCK"), 
        (1,"B", 3213,"201602", "DOCK"), 
        (1,"B", 3213,"201601", "PORT"), 
        (2,"C", 2321,"201601", "DOCK")
    ]
)
df_data = sqlContext.createDataFrame(rdd, ["id","type", "cost", "date", "ship"])

df_data.show()
 +---+----+----+------+----+
| id|type|cost|  date|ship|
+---+----+----+------+----+
|  0|   A| 223|201603|PORT|
|  0|   A|  22|201602|PORT|
|  0|   A| 422|201601|DOCK|
|  1|   B|3213|201602|DOCK|
|  1|   B|3213|201601|PORT|
|  2|   C|2321|201601|DOCK|
+---+----+----+------+----+

et j'ai besoin de pivot en date:

df_data.groupby(df_data.id, df_data.type).pivot("date").avg("cost").show()

+---+----+------+------+------+
| id|type|201601|201602|201603|
+---+----+------+------+------+
|  2|   C|2321.0|  null|  null|
|  0|   A| 422.0|  22.0| 223.0|
|  1|   B|3213.0|3213.0|  null|
+---+----+------+------+------+

Tout fonctionne comme prévu. Mais maintenant, j'ai besoin de pivot et d'obtenir un non-colonne numérique:

df_data.groupby(df_data.id, df_data.type).pivot("date").avg("ship").show()

et bien sûr, je voudrais obtenir une exception:

AnalysisException: u'"ship" is not a numeric column. Aggregation function can only be applied on a numeric column.;'

Je voudrais générer quelque chose sur la ligne de

+---+----+------+------+------+
| id|type|201601|201602|201603|
+---+----+------+------+------+
|  2|   C|DOCK  |  null|  null|
|  0|   A| DOCK |  PORT| DOCK|
|  1|   B|DOCK  |PORT  |  null|
+---+----+------+------+------+

Est-ce possible avec pivot?

OriginalL'auteur Ivan | 2016-05-27

  1. 27

    En supposant que (id |type | date) combinaisons sont uniques et votre seul but est de pivotement et de ne pas l'agrégation, vous pouvez utiliser first (ou toute autre fonction n'est pas limitée à des valeurs numériques):

    from pyspark.sql.functions import first

(df_data
    .groupby(df_data.id, df_data.type)
    .pivot("date")
    .agg(first("ship"))
    .show())

## +---+----+------+------+------+
## | id|type|201601|201602|201603|
## +---+----+------+------+------+
## |  2|   C|  DOCK|  null|  null|
## |  0|   A|  DOCK|  PORT|  PORT|
## |  1|   B|  PORT|  DOCK|  null|
## +---+----+------+------+------+

    Si ces hypothèses est pas correct, vous devrez pré-agréger vos données. Par exemple pour la plupart des ship valeur:

    from pyspark.sql.functions import max, struct

(df_data
    .groupby("id", "type", "date", "ship")
    .count()
    .groupby("id", "type")
    .pivot("date")
    .agg(max(struct("count", "ship")))
    .show())

## +---+----+--------+--------+--------+
## | id|type|  201601|  201602|  201603|
## +---+----+--------+--------+--------+
## |  2|   C|[1,DOCK]|    null|    null|
## |  0|   A|[1,DOCK]|[1,PORT]|[1,PORT]|
## |  1|   B|[1,PORT]|[1,DOCK]|    null|
## +---+----+--------+--------+--------+
    Une autre solution serait de collect_set pour garder tous les ship valeurs.
    pouvez-vous donner la solution ici
    Remplacer max(struct ci-dessus avec collect_set et vous avez terminé. La recherche de la possibilité de l'utiliser comme un acteur à part entière de réponse. Vous connaissez toutes les questions qui mendient pour une telle réponse? 😉

    OriginalL'auteur zero323