您需要一种方法来保留当前的行顺序，以便对周期进行计数。为此，可以使用函数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)

字符串