评估nans切片大小
评估nans切片大小
提问于 2019-05-27 18:50:37
我有一个spark dataframe，在列中有一些null值。我需要对非空值之前的连续null值进行计数。
使用numpy，我会这样做(代码没有针对numpy进行优化，因为我尽量不在我的问题中使用它)：
import numpy as np
x = np.array([[0, None], [1, 3.], [2, 7.], [3, None], [4, 4.], [5, 3.],
[6, None], [7, None], [8, 5.], [9, 2.], [10, None]])
def nan_count(l, n):
assert n <= len(l) + 1
assert n >= 0
if n < 1 or l[n-1] is not None:
return 0
return nan_count(l, n-1) + 1
y = map(lambda i: nan_count(x[:,1], i), x[:,0])
res = np.concatenate([x, np.asarray(y).reshape(-1,1)], axis = 1)
res因此输出如下所示：
Out[31]: [0, 1, 0, 0, 1, 0, 0, 1, 2, 0, 0]现在，如果我有一个像x这样的spark DataFrame
x = sc.parallelize([[0, None], [1, 3.], [2, 7.], [3, None], [4, 4.],
[5, 3.], [6, None], [7, None], [8, 5.], [9, 2.], [10, None]])\
.toDF()
x.show()+---+----+
| _1| _2|
+---+----+
| 0|null|
| 1| 3.0|
| 2| 7.0|
| 3|null|
| 4| 4.0|
| 5| 3.0|
| 6|null|
| 7|null|
| 8| 5.0|
| 9| 2.0|
| 10|null|
+---+----+如何获得相同的输出？
我已经尝试了一些使用udf的变通方法，但是我在引用所选择的值之前遇到了问题(我尝试在udf中使用select和filter pyspark.sql.dataframe.DataFrame方法，但这是不允许的)。
编辑：我不知道我可能会找到多少连续的nans。
回答 1
Stack Overflow用户
回答已采纳
发布于 2019-05-30 21:59:44
我将注释放入代码中来解释每个步骤，直到达到所需的输出。
当然，从下面的示例中创建所有列是没有必要的，这段代码可能会有很大的改进，但我认为很重要的一点是，一步一步地向你展示，并做一个初始的开始来解决你的问题。
x = sc.parallelize([
[0, None],
[1, 3.],
[2, 7.],
[3, None],
[4, 4.],
[5, 3.],
[6, None],
[7, None],
[8, 5.],
[9, 2.],
[10, None]
])
# Assigned values in columns A and B to facilitate manipulation
x = x.toDF(['A', 'B'])
# Prints initial DF
x.show()输出：
+---+----+
| A| B|
+---+----+
| 0|null|
| 1| 3.0|
| 2| 7.0|
| 3|null|
| 4| 4.0|
| 5| 3.0|
| 6|null|
| 7|null|
| 8| 5.0|
| 9| 2.0|
| 10|null|
+---+----+# Transform null values into "1"
x = x.withColumn('C', when(x.B.isNull(), 1))
x.show()输出：
+---+----+----+
| A| B| C|
+---+----+----+
| 0|null| 1|
| 1| 3.0|null|
| 2| 7.0|null|
| 3|null| 1|
| 4| 4.0|null|
| 5| 3.0|null|
| 6|null| 1|
| 7|null| 1|
| 8| 5.0|null|
| 9| 2.0|null|
| 10|null| 1|
+---+----+----+# Creates a spec that order column A
order_spec = Window().orderBy('A')
# Doing a cumulative sum. See the explanation
# https://stackoverflow.com/questions/56384625/pyspark-cumulative-sum-with-reset-condition
x = x \
.withColumn('tmp', sum((x.C.isNull()).cast('int')).over(order_spec)) \
.withColumn('D', sum(x.C).over(order_spec.partitionBy("tmp"))) \
.drop('tmp')
x.show()输出：
+---+----+----+----+
| A| B| C| D|
+---+----+----+----+
| 0|null| 1| 1|
| 1| 3.0|null|null|
| 2| 7.0|null|null|
| 3|null| 1| 1|
| 4| 4.0|null|null|
| 5| 3.0|null|null|
| 6|null| 1| 1|
| 7|null| 1| 2|
| 8|null| 1| 3|
| 9| 5.0|null|null|
| 10| 2.0|null|null|
| 11|null| 1| 1|
+---+----+----+----+# Put values from column D to one row above and select the desired output values
x = x.withColumn('E', lag(x.D, ).over(order_spec)) \
.select(x.A, x.B, when(col('E').isNotNull(), col('E')).otherwise(0).alias('nan_count'))
x.show()输出：
+---+----+---------+
| A| B|nan_count|
+---+----+---------+
| 0|null| 0|
| 1| 3.0| 1|
| 2| 7.0| 0|
| 3|null| 0|
| 4| 4.0| 1|
| 5| 3.0| 0|
| 6|null| 0|
| 7|null| 1|
| 8|null| 2|
| 9| 5.0| 3|
| 10| 2.0| 0|
| 11|null| 0|
+---+----+---------+完整代码：
from pyspark.shell import sc
from pyspark.sql import Window
from pyspark.sql.functions import lag, when, sum, col
x = sc.parallelize([
[0, None], [1, 3.], [2, 7.], [3, None], [4, 4.],
[5, 3.], [6, None], [7, None], [8, None], [9, 5.], [10, 2.], [11, None]])
x = x.toDF(['A', 'B'])
# Transform null values into "1"
x = x.withColumn('C', when(x.B.isNull(), 1))
# Creates a spec that order column A
order_spec = Window().orderBy('A')
# Doing a cumulative sum with reset condition. See the explanation
# https://stackoverflow.com/questions/56384625/pyspark-cumulative-sum-with-reset-condition
x = x \
.withColumn('tmp', sum((x.C.isNull()).cast('int')).over(order_spec)) \
.withColumn('D', sum(x.C).over(order_spec.partitionBy("tmp"))) \
.drop('tmp')
# Put values from column D to one row above and select the desired output values
x = x.withColumn('E', lag(x.D, ).over(order_spec)) \
.select(x.A, x.B, when(col('E').isNotNull(), col('E')).otherwise(0).alias('nan_count'))
x.show()页面原文内容由Stack Overflow提供。腾讯云小微IT领域专用引擎提供翻译支持
原文链接：
https://stackoverflow.com/questions/56324636
复制相关文章
相似问题