在PySpark中查找连续的逐月注册期

from pyspark.sql import SparkSession
from pyspark.sql.functions import lag, datediff, expr
from pyspark.sql.window import Window

spark = SparkSession.builder.appName("Continuous Monthly Registration").getOrCreate()

data = spark.read.csv("path_to_data.csv", header=True, inferSchema=True)

data = data.withColumn("registration_date", expr("to_date(registration_date, 'yyyy-MM-dd')"))

data = data.orderBy("registration_date")

data = data.withColumn("days_since_last_registration", datediff(data.registration_date, lag(data.registration_date).over(Window.orderBy("registration_date"))))

data = data.withColumn("continuous_monthly_registration", expr("CASE WHEN days_since_last_registration = 30 THEN 1 ELSE 0 END"))

data.show()