在不移动行的情况下对具有特定条件的行进行计数 （PySpark）

您需要一种方法来保留当前行顺序，以便计算周期。为此，有一个函数可用于为每行分配一个唯一的“递增值”，因此我们按该列排序，保留原始行序列。一旦到位，我们就可以使用基于窗口的计算（需要对行进行排序）来提供周期计数。monotonically_increasing_id()

注意：不确定连续 2 行的要求，下面不讨论

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

# Create SparkSession
spark = SparkSession.builder \
          .appName('CycleCount') \
          .getOrCreate()

# Assume "df" as the DataFrame
df.show()

# Create a new column "index" so that we can retain the current row order
df = df.withColumn("index", F.monotonically_increasing_id())

# Define the window ordered by the "index" which retains current row order (per letter)
window = Window.partitionBy("letter").orderBy("index")

# Add new column "change" that flags when a change in 'group' occurs
df = df.withColumn("change", F.when(F.lag("group").over(window) != F.col("group"), 1).otherwise(0))

# Use cumsum() for cumulative sum of these changes, which will serve as an identifier for each cycle
df = df.withColumn("cycle", F.sum("change").over(window))

# Filter out groups with only 1 row
df = df.filter(F.col("count") > 1)

# Count the number of cycles
cycle_count = df.groupBy("letter", "cycle").count().count()

print("Number of cycles: ", cycle_count)