PySpark连续0出现次数及分布计算：代码问题求助

原代码核心问题分析 #

1. 排序逻辑错误：原代码用Window.orderBy("Flag")会将所有相同Flag值的行聚合在一起，完全破坏了数据的原始顺序，而连续序列的统计必须基于数据的原始行顺序。Spark DataFrame本身是无序的，必须指定一个能代表原始顺序的列（如行号、时间戳等）来排序。
2. Streak标记逻辑错误：start_streak的计算式Flag*Flag - lag(Flag)完全无法正确识别连续0序列的起始点，逻辑完全偏离需求。
3. Streak分组错误：基于错误的排序窗口计算streak_id，导致无法正确划分连续的0序列。
4. 统计逻辑错误：过滤streak_id > 3的逻辑完全错误，我们需要统计的是连续0的长度超过3的序列次数，而非streak_id的值。

修正后的代码实现 #

以下代码同时完成两个需求：统计连续0长度超3的次数，以及各长度的出现分布：

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

spark = SparkSession.builder.appName("ConsecutiveZeros").getOrCreate()

# 读取数据，添加行号以保持原始顺序（CSV按行读取的顺序）
df = spark.read.csv("file.csv", header=True, inferSchema=True)
df = df.withColumn("row_num", F.monotonically_increasing_id())

# 定义窗口：按原始行顺序排序
window_order = Window.orderBy("row_num")

# 1. 标记连续0序列的起始点
df = df.withColumn(
    "prev_flag", F.lag("Flag").over(window_order)
).withColumn(
    "is_new_streak",
    when(
        (F.col("Flag") == 0) & 
        (F.col("prev_flag").isNull() | (F.col("prev_flag") != 0)),
        1
    ).otherwise(0)
)

# 2. 给每个连续0序列分配唯一ID
window_streak = Window.orderBy("row_num").rowsBetween(Window.unboundedPreceding, 0)
df = df.withColumn(
    "streak_id", F.sum("is_new_streak").over(window_streak)
)

# 3. 筛选出Flag=0的行，计算每个streak的长度
zero_streaks = df.filter(F.col("Flag") == 0).groupBy("streak_id").agg(
    F.count("*").alias("streak_length")
)

# 需求1：统计连续0长度超过3的次数
long_streak_count = zero_streaks.filter(F.col("streak_length") > 3).count()
print(f"连续0长度超过3的次数：{long_streak_count}")

# 需求2：统计连续0块长度的出现次数分布
streak_distribution = zero_streaks.groupBy("streak_length").agg(
    F.count("streak_id").alias("occurrences")
).orderBy("streak_length")
print("连续0块长度的出现分布：")
streak_distribution.show()

关键说明 #

• 行号的作用：monotonically_increasing_id()生成唯一递增的行号，确保数据保持CSV读取时的原始顺序，这是统计连续序列的基础。
• Streak起始标记：通过对比当前Flag和前一行Flag，准确识别每个连续0序列的起始位置。
• Streak分组与长度计算：通过分组streak_id得到每个连续0序列的长度，再基于长度完成两个统计需求。

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