我有一個(gè) Spark 數(shù)據(jù)框：df1 如下：age = spark.createDataFrame(["10","11","13"], "string").toDF("age")age.show()+---+|age|+---+| 10|| 11|| 13|+---+我需要向數(shù)據(jù)框中添加行號列以使其：+---+------+|age|col_id|+---+------+| 10| 1 || 11| 2 || 13| 3 |+---+------+我的數(shù)據(jù)框中的所有列都不包含唯一值。我嘗試使用F.monotonically_increasing_id())，但它只是按遞增順序生成隨機(jī)數(shù)。>>> age = spark.createDataFrame(["10","11","13"], "string").toDF("age").withColumn("rowId1", F.monotonically_increasing_id())>>> ageDataFrame[age: string, rowId1: bigint]>>> age.show<bound method DataFrame.show of DataFrame[age: string, rowId1: bigint]>>>> age.show()+---+-----------+|age| rowId1|+---+-----------+| 10|17179869184|| 11|42949672960|| 13|60129542144|+---+-----------+由于我沒有任何包含唯一數(shù)據(jù)的列，因此我擔(dān)心使用窗口函數(shù)和生成row_numbers。那么，有沒有一種方法可以row_count在數(shù)據(jù)框中添加一列，該列給出：+---+------+|age|col_id|+---+------+| 10| 1 || 11| 2 || 13| 3 |+---+------+如果窗口功能是唯一的實(shí)現(xiàn)方法，我如何確保所有數(shù)據(jù)都位于單個(gè)分區(qū)下？或者如果有一種方法可以在不使用窗口函數(shù)的情況下實(shí)現(xiàn)相同的功能，那么如何實(shí)現(xiàn)它？任何幫助表示贊賞。
2 回答
慕標(biāo)琳琳
TA貢獻(xiàn)1830條經(jīng)驗(yàn) 獲得超9個(gè)贊
使用zipWithIndex。
pyspark 與 Scala 不同。
其他答案對性能不利 - 使用單個(gè)執(zhí)行器。zipWithIndex是narrow transformation這樣，它可以按partition.
在這里，您可以進(jìn)行相應(yīng)的定制：
from pyspark.sql.types import StructField
from pyspark.sql.types import StructType
from pyspark.sql.types import StringType, LongType
import pyspark.sql.functions as F
df1 = spark.createDataFrame([ ('abc'),('2'),('3'),('4'), ('abc'),('2'),('3'),('4'), ('abc'),('2'),('3'),('4') ], StringType())
schema = StructType(df1.schema.fields[:] + [StructField("index", LongType(), True)])
rdd = df1.rdd.zipWithIndex()
rdd1 = rdd.map(lambda row: tuple(row[0].asDict()[c] for c in schema.fieldNames()[:-1]) + (row[1],))
df1 = spark.createDataFrame(rdd1, schema)
df1.show()
返回：
+-----+-----+
|value|index|
+-----+-----+
|? abc|? ? 0|
|? ? 2|? ? 1|
|? ? 3|? ? 2|
|? ? 4|? ? 3|
|? abc|? ? 4|
|? ? 2|? ? 5|
|? ? 3|? ? 6|
|? ? 4|? ? 7|
|? abc|? ? 8|
|? ? 2|? ? 9|
|? ? 3|? ?10|
|? ? 4|? ?11|
+-----+-----+
寶慕林4294392
TA貢獻(xiàn)2021條經(jīng)驗(yàn) 獲得超8個(gè)贊
假設(shè)：這個(gè)答案基于以下假設(shè)： 的順序col_id應(yīng)取決于age列。如果假設(shè)不成立，則其他建議的解決方案是問題評論中提到的zipWithIndex。zipWithIndex可以在此答案中找到 的示例用法。
建議的解決方案：您可以使用window帶有空partitionBy和行號的 a 來獲取預(yù)期的數(shù)字。
from pyspark.sql.window import Window
from pyspark.sql import functions as F
windowSpec = Window.partitionBy().orderBy(F.col('age').asc())
age = age.withColumn(
'col_id',
F.row_number().over(windowSpec)
)
添加回答
舉報(bào)
0/150
提交
取消