Почему мой столбец row_number в PySpark испорчен при применении схемы? ⇐ Python

[Anonymous]

Я хочу применить схему к определенным нетехническим столбцам кадра данных Spark. Предварительно я добавляю искусственный идентификатор, используя Window и row_number, чтобы позже можно было присоединиться к новому DataFrame некоторых других технических столбцов из исходного DataFrame. Однако после применения схемы сгенерированный идентификатор искажается. Ниже приведен пример кода. Может кто-нибудь объяснить, почему это происходит и как решить проблему?
from pyspark.sql.functions import row_number, lit, col, monotonically_increasing_id, sum
from pyspark.sql.window import Window
from pyspark.sql.types import StructType, StructField, IntegerType, StringType

# Sample DataFrame
data = [(1, "Alice"), (2, "Bob"), (3, "Charlie")]
df = spark.createDataFrame(data, ["id", "name"])

# Schema to apply
schema = StructType([
StructField("id", IntegerType(), False),
StructField("name", StringType(), False),
])

# Create ID column
w = Window().orderBy(lit('A'))
df = df.withColumn('_special_surrogate_id', row_number().over(w))

# Improved method
surrogate_key_field = StructField("_special_surrogate_id", StringType(), False)
schema_with_surrogate = StructType(schema.fields + [surrogate_key_field])

# Loop because sometimes it works and sometimes it does't work
for i in range(11):

df_filtered = df.select("id", "name", "_special_surrogate_id")
df_filtered = spark.createDataFrame(df_filtered.rdd, schema_with_surrogate)

combined_df = df.withColumnRenamed("id", "id1").join(df_filtered.withColumnRenamed("id", "id2"), on="_special_surrogate_id")

print("Diffs in Iteration " + str(i) + ":")
print(combined_df.withColumn("diff", (col("id1") != col("id2")).cast("integer")).agg(sum("diff")).collect()[0][0])



Подробнее здесь: https://stackoverflow.com/questions/786 ... g-a-schema