"Spalte ist nicht iterbar", wenn Operationen mit DataFrame als Teil der Funktion durchgeführt werden ⇐ Python

[Guest]

Guten Tag, alle. Ich versuche, einen Vergleich mit der Berechnung zwischen zwei Datenrahmen durchzuführen und eine Ausgabe in einer Tabelle mit zwei Spalten zu geben: "Art der Änderung" und "Anzahl der Vorschläge". Dazu habe ich zwei Datenrahmen zusammengeführt und "_New" und "_old" -Suffixe zu Namen von Spalten hinzugefügt, um sie zu identifizieren, und schrieb eine kleine Funktion, um sie zu vergleichen, zu summieren und eine Ausgabe in Form von DataFrame zu geben. Der Code sieht so aus: < /p>

Code: Select all

# Loop to go through columns and count changes
def Form_Change_Report(
ColumnList,
Comparison_Dataframe: Comparison_DF,
ChangeReport_Dataframe: ChangeReport_DF,
):
for x in ColumnList:
Comparison_Dataframe = Comparison_Dataframe.withColumn(
x + "_Change", when(col(x + "_Old") == col(x + "_New"), 0).otherwise(1)
)
Change_DF = Comparison_Dataframe.select(
lit("Change to " + x).alias("Type of Change"),
sum(x + "_Change").alias("Number of Occurences"),
)
ChangeReport_Dataframe = ChangeReport_Dataframe.unionByName(Change_DF)

return ChangeReport_Dataframe

# Forming blank dataframe for change report
ChangeReport_RDD = spark.sparkContext.emptyRDD()
ChangeReport_Columns = [
StructField("Type of Change", StringType(), True),
StructField("Number of Occurences", IntegerType(), True),
]
ChangeReport_DF = spark.createDataFrame([], schema=StructType(ChangeReport_Columns))

ChangeReport_DF = Form_Change_Report(Columns_To_Compare, Comparison_DF, ChangeReport_DF)
< /code>
Beim Versuch, sie auszuführen, gibt die Fehler "Spalte in nicht iterierbar" zurück. Ich habe versucht, Net auf der Suche nach Lösung zu durchsuchen, aber alle Beispiele, die ich fand, basierten auf der Verwendung der MAX -Funktion, was in diesem Szenario nicht der Fall ist. Dies ist notwendige Information
würde jemand darauf hinweisen, was das [url=viewtopic.php?t=11587]Problem[/url] sein könnte? um einen Reprex hinzuzufügen.  Code, den ich so aussieht, sieht so aus: < /p>
import pyspark.sql.functions as f
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, when
from pyspark.sql.functions import sum as fsum
from pyspark.sql.types import StructType, StructField

data1 = [("Postcode1", "100","150"),("Postcode2", "200","250"),("Postcode3", "300","350"),("Postcode4", "400","450")]
data2 = [("Postcode1", "150","150"),("Postcode2", "200","200"),("Postcode3", "350","350"),("Postcode4", "400","450")]
Columns = ["Postcode", "Count1","Count2"]

rdd1 = spark.sparkContext.parallelize(data1)
rdd2 = spark.sparkContext.parallelize(data2)

df1 = spark.createDataFrame(rdd1, schema=Columns)
df2 = spark.createDataFrame(rdd2, schema=Columns)

ColumnNames_New =[
f.col("Postcode"),
f.col("Count1").alias("Count1_New"),
f.col("Count2").alias("Count2_New")
]

df1_NewData = df1.select(ColumnNames_New)

ColumnNames_Old =[
f.col("Postcode"),
f.col("Count1").alias("Count1_Old"),
f.col("Count2").alias("Count2_Old")
]

df2_OldData = df2.select(ColumnNames_Old)

#Joining two dataframes on postcode
Comparison_DF=df1_NewData.join(df2_OldData, "Postcode")

Columns_to_Compare = [
f.col("Count1"),
f.col("Count2")
]

#Forming blank dataframe for change report
ChangeReport_RDD=spark.sparkContext.emptyRDD()
ChangeReport_Columns=[
StructField("Type of Change",StringType(),True),
StructField("Number of Occurences",IntegerType(),True)
]
ChangeReport_DF=spark.createDataFrame([], schema=StructType(ChangeReport_Columns))

def Form_Change_Report(ColumnList,Comparison_Dataframe:df1_NewData, ChangeReport_Dataframe:ChangeReport_DF):
for x in ColumnList:
Comparison_Dataframe=Comparison_Dataframe.withColumn(x+"_Change",when(col(x+"_Old")==col(x+"_New"),0).otherwise(1))
Change_DF = Comparison_Dataframe.select(lit("Change to "+x).alias("Type of Change"),fsum(x+"_Change").alias("Number of Occurences"))
ChangeReport_Dataframe=ChangeReport_Dataframe.unionByName(Change_DF)

return ChangeReport_Dataframe

ChangeReport_DF=Form_Change_Report(Columns_to_Compare,Comparison_DF,ChangeReport_DF