如何在pyspark中找到变异系数（CV）？答案

【问题标题】：How to find coefficient of variation (CV) in pyspark?如何在pyspark中找到变异系数（CV）？
【发布时间】：2019-08-28 11:55:29
【问题描述】：

我有 2 个 pyspark 数据帧，我想找到这两个数据帧的变异系数。

数据框1：

          hours    total

           00     75969.0
           01     75302.0
           02     74636.0
           03     73969.0
           04     73302.0
           05     72635.0

数据框2：-

            hours   total1

             00      71535
             01      71182
             02      77628
             03      75984
             04      75276
             05      67259

我想要这样的输出：-

dataframe3 :-

      hours       total        total1   CV
       00        75969.0        71535   3.006020
       01        75302.0        71182   2.812594
       02        74636.0        77628   1.965008
       03        73969.0        75984   1.343754
       04        73302.0        75276   1.328595
       05        72635.0        67259   3.842910

我通过将 pyspark-dataframe 转换为 pandas 数据帧来完成这些，但我想在不使用 pandas 的情况下计算 CV。这些我都做过

     pd1=dataframe1.toPandas()
     pd2=dataframe2.toPandas()
     a4=[]
     list1=[]
     count=len(pd1)  
     print(count)
     import numpy as np
     for i in range(count):
         del a4[:]
         p9=(pd1.total[i])
         p10=(pd2.total1[i])
         a4.append(p10)
         a4.append(p9)
         standard_d1=np.std(a4,ddof=0)
         mean1=np.mean(a4)
         cv=(standard_d1/mean1)*100
         list1.append(cv)
     pd1['cv']=list1

【问题讨论】：

标签： pyspark apache-spark-sql

【解决方案1】：

有几种方法可以解决这个问题：

使用window
使用udf
使用window + udf
- 首先，让我们构建 DataFrame：

from pyspark.sql.window import Window
from pyspark.sql import functions as F
from pyspark.sql.types import *
from pyspark.sql.functions import mean, pandas_udf, PandasUDFType


a1 = [(0, 75969.0), (1, 75302.0), (2, 74636.0), (3, 73969.0), (4, 73302.0), (5, 72635.0)]
a2 = [(0, 71535.0), (1, 71182.0), (2, 77628.0), (3, 75984.0), (4, 75276.0), (5, 67259.0)]
df1 = spark.createDataFrame(a1, ['hours', 'total'])
df2 = spark.createDataFrame(a2, ['hours', 'total'])
df = df1.union(df2)
df.show()

+-----+-------+
|hours|  total|
+-----+-------+
|    0|75969.0|
|    1|75302.0|
|    2|74636.0|
|    3|73969.0|
|    4|73302.0|
|    5|72635.0|
|    0|71535.0|
|    1|71182.0|
|    2|77628.0|
|    3|75984.0|
|    4|75276.0|
|    5|67259.0|
+-----+-------+

仅使用udf


@pandas_udf(FloatType(), PandasUDFType.GROUPED_AGG)
def _udf(v):
    return 100.0*np.std(v, ddof=0)/np.mean(v)

df = df.groupBy('hours').agg(_udf(df['total']).alias('CV')).orderBy('hours')
df.show()

使用window

w = Window.partitionBy('hours')
df = df.withColumn('std', F.stddev_pop('total').over(w))
df = df.withColumn('mean', F.mean('total').over(w))
df = df.withColumn('CV', 100.0*df['std']/df['mean']).dropDuplicates(['hours']).drop(*['total', 'std', 'mean']).orderBy('hours')
df.show()

使用window + udf

w = Window.partitionBy('hours')
df = df.withColumn('CV',_udf('total').over(w)).dropDuplicates(['hours']).orderBy('hours')
df.show()

以上所有方法都给你结果：

+-----+---------+
|hours|       CV|
+-----+---------+
|    0|  3.00602|
|    1| 2.812594|
|    2|1.9650081|
|    3|1.3437544|
|    4| 1.328595|
|    5|3.8429096|
+-----+---------+

【讨论】：

感谢您的宝贵时间，但它的计算时间更多..@niuer

【解决方案2】：

因为你只有两个元素，所以我们有 standard_dev = |total - total1| / 2 的属性。因为mean = (total + total1) / 2，我们有CV = 100 * |total - total1| / (total + total1)。

因此：

from pyspark.sql import functions as F
df = dataframe1.join(dataframe2, 'hours', 'inner')
df_final = df.withColumn('CV', F.lit(100) * F.abs(df['total'] - df['total1']) / (df['total'] + df['total1']))

【讨论】：