在 PySpark Dataframe 中为 n 行添加一个关键元素
Add a key element for n rows in PySpark Dataframe
我有一个如下所示的数据框。
id | run_id
--------------
4 | 12345
6 | 12567
10 | 12890
13 | 12450
我想添加一个新列 Key，前 n 行的值为 1，接下来的 n 行的值为 2。结果将是这样的：
id | run_id | key
----------------------
4 | 12345 | 1
6 | 12567 | 1
10 | 12890 | 2
13 | 12450 | 2
是否可以对 PySpark 做同样的事情？在此先感谢您的帮助。
这是使用 zipWithIndex 的一种方法：
# sample rdd
rdd=sc.parallelize([[4,12345], [6,12567], [10,12890], [13,12450]])
# group size for key
n=2
# add rownumber and then label in batches of size n
rdd=rdd.zipWithIndex().map(lambda (x, rownum): x+[int(rownum/n)+1])
# convert to dataframe
df=rdd.toDF(schema=['id', 'run_id', 'key'])
df.show(4)
我有一个如下所示的数据框。
id | run_id
--------------
4 | 12345
6 | 12567
10 | 12890
13 | 12450
我想添加一个新列 Key，前 n 行的值为 1，接下来的 n 行的值为 2。结果将是这样的：
id | run_id | key
----------------------
4 | 12345 | 1
6 | 12567 | 1
10 | 12890 | 2
13 | 12450 | 2
是否可以对 PySpark 做同样的事情？在此先感谢您的帮助。
这是使用 zipWithIndex 的一种方法：
# sample rdd
rdd=sc.parallelize([[4,12345], [6,12567], [10,12890], [13,12450]])
# group size for key
n=2
# add rownumber and then label in batches of size n
rdd=rdd.zipWithIndex().map(lambda (x, rownum): x+[int(rownum/n)+1])
# convert to dataframe
df=rdd.toDF(schema=['id', 'run_id', 'key'])
df.show(4)