Pyspark/Hive 中带条件的加权运行总计

Question

我有产品、品牌、百分比和价格列。我想计算与当前行具有不同品牌的行以及与当前行具有相同品牌的行的百分比列的总和。我想按价格衡量它们。如果当前行上方的产品价格高于当前行，我想通过将其乘以 0.8 来降低权重。如何在 PySpark 中或使用 spark.sql 中执行此操作？不使用乘以权重的答案是here。

df = pd.DataFrame({'a': ['a1','a2','a3','a4','a5','a6'],
          'brand':['b1','b2','b1', 'b3', 'b2','b1'],
      'pct': [40, 30, 10, 8,7,5], 
       'price':[0.6, 1, 0.5, 0.8, 1, 0.5]})
df = spark.createDataFrame(df)

我在寻找什么

product  brand  pct  pct_same_brand  pct_different_brand
a1       b1     40     null            null
a2       b2     30     null            40
a3       b1     10     32              30
a4       b3     8      null            80
a5       b2     7      24              58
a6       b1     5      40              45  

更新： 我添加了以下数据点以帮助澄清问题。可以看出，一行可以在一行中乘以 0.8，在另一行中乘以 1.0。

  product   brand   pct  price  pct_same_brand  pct_different_brand
  a1        b1      30   0.6    null                 null
  a2        b2      20   1.3    null                 30
  a3        b1      10   0.5    30*0.8               20
  a4        b3      8    0.8    null                 60
  a5        b2      7    0.5    20*0.8               48
  a6        b1      6    0.8    30*1 + 10*1          35
  a7        b2      5    1.5    20*1 + 7*1           54

Update2：在我上面提供的数据中，每行的权重是相同的数字（0.8 或 1），但也可以是 1 和 0.8（某些行为 0.8，而 1对于其他行）

以下数据框中的示例，例如，最后一行的乘数应该是 a6 的 0.8 和品牌 b1 的其余部分的 1.0。 ：

  df = pd.DataFrame({'a': ['a1','a2','a3','a4','a5','a6', 'a7', 'a8', 'a9', 'a10'],
          'brand':['b1','b2','b1', 'b3', 'b2','b1','b2', 'b1', 'b1', 'b1'],
          'pct': [30, 20, 10, 8, 7,6,5,4,3,2],
          'price':[0.6, 1.3, 0.5, 0.8, 0.5, 0.8, 1.5, 0.5, 0.65, 0.7]
         
         })
df = spark.createDataFrame(df)

Answer 1

可以添加weight列方便计算：

from pyspark.sql import functions as F, Window

df2 = df.withColumn(
    'weight', 
    F.when(
        F.col('price') <= F.lag('price').over(
            Window.partitionBy('brand')
                  .orderBy(F.desc('pct'))
        ), 
        0.8
    ).otherwise(1.0)
).withColumn(
    'pct_same_brand', 
    F.col('weight')*F.sum('pct').over(
        Window.partitionBy('brand')
              .orderBy(F.desc('pct'))
              .rowsBetween(Window.unboundedPreceding, -1)
    )
).withColumn(
    'pct_different_brand', 
    F.sum('pct').over(
        Window.orderBy(F.desc('pct'))
              .rowsBetween(Window.unboundedPreceding, -1)
    ) - F.coalesce(F.col('pct_same_brand'), F.lit(0)) / F.col('weight')
)

df2.show()
+---+-----+---+-----+------+--------------+-------------------+
|  a|brand|pct|price|weight|pct_same_brand|pct_different_brand|
+---+-----+---+-----+------+--------------+-------------------+
| a1|   b1| 40|  0.6|   1.0|          null|               null|
| a2|   b2| 30|  1.0|   1.0|          null|               40.0|
| a3|   b1| 10|  0.5|   0.8|          32.0|               30.0|
| a4|   b3|  8|  0.8|   1.0|          null|               80.0|
| a5|   b2|  7|  1.0|   0.8|          24.0|               58.0|
| a6|   b1|  5|  0.5|   0.8|          40.0|               45.0|
+---+-----+---+-----+------+--------------+-------------------+

---

已编辑问题的输出：

+---+-----+---+-----+------+--------------+-------------------+
|  a|brand|pct|price|weight|pct_same_brand|pct_different_brand|
+---+-----+---+-----+------+--------------+-------------------+
| a1|   b1| 30|  0.6|   1.0|          null|               null|
| a2|   b2| 20|  1.3|   1.0|          null|               30.0|
| a3|   b1| 10|  0.5|   0.8|          24.0|               20.0|
| a4|   b3|  8|  0.8|   1.0|          null|               60.0|
| a5|   b2|  7|  0.5|   0.8|          16.0|               48.0|
| a6|   b1|  6|  0.8|   1.0|          40.0|               35.0|
| a7|   b2|  5|  1.5|   1.0|          27.0|               54.0|
+---+-----+---+-----+------+--------------+-------------------+

Answer 2

如果有人有类似的问题，这对我有用。 基本上，我使用了数据框的外连接并分配了权重。最后，我使用了窗口函数。

df_copy = df.withColumnRenamed('a', 'asin')\
        .withColumnRenamed('brand', 'brandd')\
         .withColumnRenamed('pct', 'pct2')\
         .withColumnRenamed('price', 'price2')
 df2 = df.join(df_copy, on = [df.brand == df_copy.brandd], how = 'outer').orderBy('brand')
 df3 = df2.filter(~((df2.a == df2.asin) & (df2.brand == df2.brandd))
             & (df2.pct <= df2.pct2))
 df3 = df3.withColumn('weight', F.when(df3.price2 > df3.price, 0.8).otherwise(1))

 df4 = df3.groupBy(['a', 'brand', 'pct', 'price']).agg(F.sum(df3.pct2*df3.weight)
                                        .alias('same_brand_pct'))

 df5 = df.join(df4, on = ['a', 'brand', 'pct', 'price'], how = 'left')
 df6 = df5.withColumn(
   'pct_same_brand_unscaled', 
    F.sum('pct').over(
    Window.partitionBy('brand')
          .orderBy(F.desc('pct'))
          .rowsBetween(Window.unboundedPreceding, -1)
  )
  ).withColumn(
'pct_different_brand', 
F.sum('pct').over(
    Window.orderBy(F.desc('pct'))
          .rowsBetween(Window.unboundedPreceding, -1)
  ) - F.coalesce(F.col('pct_same_brand_unscaled'), F.lit(0))
 ).drop('pct_same_brand_unscaled')

给予：

+---+-----+---+-----+--------------+-------------------+
|  a|brand|pct|price|same_brand_pct|pct_different_brand|
+---+-----+---+-----+--------------+-------------------+
| a1|   b1| 30|  0.6|          null|               null|
| a2|   b2| 20|  1.3|          null|                 30|
| a3|   b1| 10|  0.5|          24.0|                 20|
| a4|   b3|  8|  0.8|          null|                 60|
| a5|   b2|  7|  0.5|          16.0|                 48|
| a6|   b1|  6|  0.8|          40.0|                 35|
| a7|   b2|  5|  1.5|          27.0|                 54|
| a8|   b1|  4|  0.5|          38.8|                 40|
| a9|   b1|  3| 0.65|          48.8|                 40|
|a10|   b1|  2|  0.7|          51.8|                 40|```