我有一个用户日志,我从csv获取并转换为DataFrame,以利用SparkSQL查询功能.单个用户每小时会创建大量条目,我想为每个用户收集一些基本的统计信息; 实际上只是用户实例的数量,平均值以及众多列的标准偏差.我能够通过使用groupBy($"user")以及用于count和avg的SparkSQL函数的聚合器来快速获取均值和计数信息:
val meanData = selectedData.groupBy($"user").agg(count($"logOn"),
avg($"transaction"), avg($"submit"), avg($"submitsPerHour"), avg($"replies"),
avg($"repliesPerHour"), avg($"duration"))
但是,我似乎无法找到同样优雅的方法来计算标准偏差.到目前为止,我只能通过映射字符串,双对并使用StatCounter()来计算它.stdev实用程序:
val stdevduration = duration.groupByKey().mapValues(value =>
org.apache.spark.util.StatCounter(value).stdev)
然而,这会返回一个RDD,我想尝试将其全部保存在DataFrame中,以便对返回的数据进行进一步的查询.
我试图通过减去均值并除以每列的stddev来标准化spark数据帧中多列的值。这是我到目前为止的代码：
from pyspark.sql import Row
from pyspark.sql.functions import stddev_pop, avg
df = spark.createDataFrame([Row(A=1, B=6), Row(A=2, B=7), Row(A=3, B=8),
Row(A=4, B=9), Row(A=5, B=10)])
exprs = [x - (avg(x)) / stddev_pop(x) for x in df.columns]
df.select(exprs).show()
这给了我结果：
+------------------------------+------------------------------+
|(A - (avg(A) / stddev_pop(A)))|(B - (avg(B) / stddev_pop(B)))|
+------------------------------+------------------------------+
| null| null|
+------------------------------+------------------------------+
我希望的地方：
+------------------------------+------------------------------+
|(A - (avg(A) / stddev_pop(A)))|(B - (avg(B) / stddev_pop(B)))|
+------------------------------+------------------------------+
| -1.414213562| -1.414213562|
| -0.707106781| -0.707106781|
| 0| 0|
| 0.707106781| 0.707106781|
| 1.414213562| 1.414213562|
+------------------------------+------------------------------+
我相信我可以使用mllib中的StandardScaler …