我有一個 Spark 數據框：df1 如下：age = spark.createDataFrame(["10","11","13"], "string").toDF("age")age.show()+---+|age|+---+| 10|| 11|| 13|+---+我需要向數據框中添加行號列以使其：+---+------+|age|col_id|+---+------+| 10| 1 || 11| 2 || 13| 3 |+---+------+我的數據框中的所有列都不包含唯一值。我嘗試使用F.monotonically_increasing_id())，但它只是按遞增順序生成隨機數。>>> 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|+---+-----------+由于我沒有任何包含唯一數據的列，因此我擔心使用窗口函數和生成row_numbers。那么，有沒有一種方法可以row_count在數據框中添加一列，該列給出：+---+------+|age|col_id|+---+------+| 10| 1 || 11| 2 || 13| 3 |+---+------+如果窗口功能是唯一的實現方法，我如何確保所有數據都位于單個分區下？或者如果有一種方法可以在不使用窗口函數的情況下實現相同的功能，那么如何實現它？任何幫助表示贊賞。
2 回答
慕標琳琳
TA貢獻1830條經驗 獲得超9個贊
使用zipWithIndex。
pyspark 與 Scala 不同。
其他答案對性能不利 - 使用單個執行器。zipWithIndex是narrow transformation這樣，它可以按partition.
在這里，您可以進行相應的定制：
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貢獻2021條經驗 獲得超8個贊
假設：這個答案基于以下假設： 的順序col_id應取決于age列。如果假設不成立，則其他建議的解決方案是問題評論中提到的zipWithIndex。zipWithIndex可以在此答案中找到 的示例用法。
建議的解決方案：您可以使用window帶有空partitionBy和行號的 a 來獲取預期的數字。
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)
)
添加回答
舉報
0/150
提交
取消