如何将PySpark DataFrame中的'Total'行移至最后一行

原始DataFrame #

尝试过的方法 #

方法1：使用monotonically_increasing_id #

df = (df.withColumn("Id", monotonically_increasing_id()+1)
           )

方法2：结合窗口函数生成行号 #

from pyspark.sql.window import Window as W
from pyspark.sql import functions as F

df1 = df.withColumn("Id", F.monotonically_increasing_id()+1)
windowSpec = W.orderBy("Id")
df1 = df1.withColumn("Id", F.row_number().over(windowSpec)).show()

方法3：拆分合并DataFrame #

from pyspark.sql import functions as F

total_row = df.filter(F.col("Column A") == "Total")
other_rows = df.filter(F.col("Column A") != "Total")

final_df = other_rows.union(total_row)
final_df.show()

方法4：标记排序法 #

from pyspark.sql import functions as F
from pyspark.sql.window import Window as W

df1 = df1.withColumn("Flag", F.when(F.col("Column A") == "Total", 1).otherwise(0))
windowSpec = W.orderBy("Flag", "Column A")
df1 = df1.withColumn("Id", F.row_number().over(windowSpec))
df1 = df1.drop("Flag")
df1.show()

遇到的问题 #

上述方法中，拆分合并和标记排序法能将'Total'行移到末尾，但会导致其他行的Balance和Paid Amount值异常，仅Total行的计算结果保留，其他行数据被错误修改。

期望结果 #

解决方案 #

要实现将'Total'行移至末尾且保证所有数据正确，核心是避免修改原始数据行的内容，仅调整排序逻辑或安全合并：

方案1：安全拆分合并（推荐） #

拆分时严格保留原始数据，合并时确保列顺序一致，且不修改原始行的任何值：

from pyspark.sql import functions as F
from pyspark.sql.window import Window

# 拆分数据：分离Total行和其他行
total_rows = df.filter(F.col("Column A") == "Total")
non_total_rows = df.filter(F.col("Column A") != "Total")

# 合并：先放非Total行，再追加Total行（用unionByName保证列顺序一致）
final_df = non_total_rows.unionByName(total_rows)

# 按需生成连续的Id列，不影响其他数据
final_df = final_df.withColumn("Id", F.row_number().over(Window.orderBy(F.lit(1))))
final_df.show()

方案2：排序标记法（不修改原始数据） #

通过标记列控制排序，同时避免窗口函数误修改其他列：

from pyspark.sql import functions as F
from pyspark.sql.window import Window

# 添加排序标记：Total行标记为1，其他为0，排序时0在前1在后
df_with_flag = df.withColumn("sort_flag", F.when(F.col("Column A") == "Total", 1).otherwise(0))

# 按标记排序，同时保留非Total行的原始顺序，然后生成新Id
window_spec = Window.orderBy("sort_flag", "Id")
final_df = df_with_flag.withColumn("Id", F.row_number().over(window_spec)) \
                       .drop("sort_flag")

final_df.show()

关键注意点 #

• 使用unionByName而非union：确保合并时列顺序一致，避免因列顺序问题导致数据错位
• 生成新Id时，仅对Id列进行操作，不修改Balance和Paid Amount的原始值
• 如果需要保留非Total行的原始顺序，排序时加入原始Id或其他唯一标识列

内容的提问来源于stack exchange，提问作者kimchigirl