Diviser Pyspark Dataframe colonne par Colonne, en d'autres Pyspark Dataframe lors de l'ID Correspond à

J'ai un PySpark DataFrame, df1, qui ressemble à ceci:

CustomerID  CustomerValue
12          .17
14          .15
14          .25
17          .50
17          .01
17          .35

J'ai un deuxième PySpark DataFrame, df2, c'est-à df1 regroupées par code client et agrégées par la fonction somme. Il ressemble à ceci:

 CustomerID  CustomerValueSum
 12          .17
 14          .40
 17          .86

Je veux ajouter une troisième colonne à df1 qui est df1['CustomerValue'] divisé par df2['CustomerValueSum'] pour la même Customerid. Cela ressemblerait à:

CustomerID  CustomerValue  NormalizedCustomerValue
12          .17            1.00
14          .15            .38
14          .25            .62
17          .50            .58
17          .01            .01
17          .35            .41

En d'autres termes, je suis en train de convertir ce Python/Pandas code de PySpark:

normalized_list = []
for idx, row in df1.iterrows():
    (
        normalized_list
        .append(
            row.CustomerValue / df2[df2.CustomerID == row.CustomerID].CustomerValueSum
        )
    )
df1['NormalizedCustomerValue'] = [val.values[0] for val in normalized_list]

Comment puis-je faire cela?

OriginalL'auteur TrentWoodbury | 2017-04-07

  1. 7

    Code:

    import pyspark.sql.functions as F

df1 = df1\
    .join(df2, "CustomerID")\
    .withColumn("NormalizedCustomerValue", (F.col("CustomerValue") / F.col("CustomerValueSum")))\
    .drop("CustomerValueSum")

    De sortie:

    df1.show()

+----------+-------------+-----------------------+
|CustomerID|CustomerValue|NormalizedCustomerValue|
+----------+-------------+-----------------------+
|        17|          0.5|     0.5813953488372093|
|        17|         0.01|   0.011627906976744186|
|        17|         0.35|     0.4069767441860465|
|        12|         0.17|                    1.0|
|        14|         0.15|    0.37499999999999994|
|        14|         0.25|                  0.625|
+----------+-------------+-----------------------+

    OriginalL'auteur dfernig

  2. 3

    Cela peut également être réalisé à l'aide de l'Étincelle fonction de Fenêtre où vous n'avez pas besoin de créer des dataframe avec les valeurs agrégées (df2):

    La création des données pour l'entrée dataframe:

    from pyspark.sql import HiveContext
sqlContext = HiveContext(sc)

data =[(12, 0.17), (14, 0.15), (14, 0.25), (17, 0.5), (17, 0.01), (17, 0.35)]
df1 = sqlContext.createDataFrame(data, ['CustomerID', 'CustomerValue'])
df1.show()
+----------+-------------+
|CustomerID|CustomerValue|
+----------+-------------+
|        12|         0.17|
|        14|         0.15|
|        14|         0.25|
|        17|          0.5|
|        17|         0.01|
|        17|         0.35|
+----------+-------------+

    Définissant une Fenêtre partitionné par code client:

    from pyspark.sql import Window
from pyspark.sql.functions import sum

w = Window.partitionBy('CustomerID')

df2 = df1.withColumn('NormalizedCustomerValue', df1.CustomerValue/sum(df1.CustomerValue).over(w)).orderBy('CustomerID')

df2.show()
+----------+-------------+-----------------------+
|CustomerID|CustomerValue|NormalizedCustomerValue|
+----------+-------------+-----------------------+
|        12|         0.17|                    1.0|
|        14|         0.15|    0.37499999999999994|
|        14|         0.25|                  0.625|
|        17|          0.5|     0.5813953488372093|
|        17|         0.01|   0.011627906976744186|
|        17|         0.35|     0.4069767441860465|
+----------+-------------+-----------------------+

    OriginalL'auteur Abhishek Bansal