我很难理解为什么我不能运行一个转换，在等待了这么多分钟(有时是几个小时(后，它会返回错误"序列化结果过大"；。
在转换中，我有一个日期列表，我正在for循环中迭代这些日期，以便在特定的时间间隔内进行增量计算。
期望的数据集是迭代数据集的并集，应该包含450k行，不太多，但我有很多计算阶段、任务和尝试
配置文件已设置为Medium配置文件，我无法在其他配置文件上缩放，也无法设置maxResultSize=0。
代码示例：
Date_list = [All weeks from: '2021-01-01', to: '2022-01-01'] --> ~50 elements
df_total = spark.createDataframe([], schema)
df_date = []
for date in Date_list:
tmp = df.filter(between [date, date-7days]).withColumn('example', F.lit(date))
........
df2 = df.join(tmp, 'column', 'inner').......
df_date += [df2]
df_total = df_total.unionByName(union_many(*df_date))
return df_total
不要注意语法。这只是一个例子，表明循环中有一系列操作我的无序输出是一个数据帧，它包含每个迭代的数据帧
谢谢！！
初始理论
您正在达到Spark的一个已知限制，类似于这里讨论的发现。
然而，有一些方法可以解决这个问题，方法是重新思考您的实现，将其改为一系列描述您希望操作的数据批次的调度指令，类似于创建tmpDataFrame的方式。
不幸的是，这可能需要更多的工作来以这种方式重新思考您的逻辑，因为您希望将您的操作纯粹想象为一系列提供给PySpark的列操作命令，而不是逐行操作。有些操作不能完全使用PySpark调用来完成，所以这并不总是可能的。总的来说，值得仔细考虑。
具体来说
例如，您的数据范围计算可以完全在PySpark中执行，如果您在多年内或其他规模增加的情况下执行此操作，则速度会大大加快。我们不使用Python列表理解或其他逻辑，而是对一小部分初始数据使用列操作来构建我们的范围。
我在这里写了一些关于如何创建日期批次的示例代码，这应该允许您执行join来创建tmpDataFrame，之后您可以描述您希望对其执行的操作类型。
创建日期范围的代码(一年中每周的开始和结束日期(：
from pyspark.sql import types as T, functions as F, SparkSession, Window
from datetime import date
spark = SparkSession.builder.getOrCreate()
year_marker_schema = T.StructType([
T.StructField("max_year", T.IntegerType(), False),
])
year_marker_data = [
{"max_year": 2022}
]
year_marker_df = spark.createDataFrame(year_marker_data, year_marker_schema)
year_marker_df.show()
"""
+--------+
|max_year|
+--------+
| 2022|
+--------+
"""
previous_week_window = Window.partitionBy(F.col("start_year")).orderBy("start_week_index")
year_marker_df = year_marker_df.select(
(F.col("max_year") - 1).alias("start_year"),
"*"
).select(
F.to_date(F.col("max_year").cast(T.StringType()), "yyyy").alias("max_year_date"),
F.to_date(F.col("start_year").cast(T.StringType()), "yyyy").alias("start_year_date"),
"*"
).select(
F.datediff(F.col("max_year_date"), F.col("start_year_date")).alias("days_between"),
"*"
).select(
F.floor(F.col("days_between") / 7).alias("weeks_between"),
"*"
).select(
F.sequence(F.lit(0), F.col("weeks_between")).alias("week_indices"),
"*"
).select(
F.explode(F.col("week_indices")).alias("start_week_index"),
"*"
).select(
F.lead(F.col("start_week_index"), 1).over(previous_week_window).alias("end_week_index"),
"*"
).select(
((F.col("start_week_index") * 7) + 1).alias("start_day"),
((F.col("end_week_index") * 7) + 1).alias("end_day"),
"*"
).select(
F.concat_ws(
"-",
F.col("start_year"),
F.col("start_day").cast(T.StringType())
).alias("start_day_string"),
F.concat_ws(
"-",
F.col("start_year"),
F.col("end_day").cast(T.StringType())
).alias("end_day_string"),
"*"
).select(
F.to_date(
F.col("start_day_string"),
"yyyy-D"
).alias("start_date"),
F.to_date(
F.col("end_day_string"),
"yyyy-D"
).alias("end_date"),
"*"
)
year_marker_df.drop(
"max_year",
"start_year",
"weeks_between",
"days_between",
"week_indices",
"max_year_date",
"start_day_string",
"end_day_string",
"start_day",
"end_day",
"start_week_index",
"end_week_index",
"start_year_date"
).show()
"""
+----------+----------+
|start_date| end_date|
+----------+----------+
|2021-01-01|2021-01-08|
|2021-01-08|2021-01-15|
|2021-01-15|2021-01-22|
|2021-01-22|2021-01-29|
|2021-01-29|2021-02-05|
|2021-02-05|2021-02-12|
|2021-02-12|2021-02-19|
|2021-02-19|2021-02-26|
|2021-02-26|2021-03-05|
|2021-03-05|2021-03-12|
|2021-03-12|2021-03-19|
|2021-03-19|2021-03-26|
|2021-03-26|2021-04-02|
|2021-04-02|2021-04-09|
|2021-04-09|2021-04-16|
|2021-04-16|2021-04-23|
|2021-04-23|2021-04-30|
|2021-04-30|2021-05-07|
|2021-05-07|2021-05-14|
|2021-05-14|2021-05-21|
+----------+----------+
only showing top 20 rows
"""
潜在的优化
一旦您有了这段代码，并且如果您无法单独通过联接/列派生来表达您的工作，并且被迫使用union_many执行操作，您可以考虑在df2结果上使用Spark的localCheckpoint功能。这将允许Spark简单地计算结果DataFrame，而不将其查询计划添加到您将推送到df_total的结果中。这可以与缓存配对，也可以将生成的DataFrame保存在内存中，但这将取决于您的数据规模。
localCheckpoint和cache有助于避免多次重新计算相同的DataFrames，并截断在中间DataFrames之上完成的查询计划量。
您可能会发现localCheckpoint和cache在dfDataFrame上也很有用，因为它将在循环中多次使用(假设您无法重新调整逻辑以使用基于SQL的操作，而是仍然被迫使用循环(。
作为何时使用每个的快速而肮脏的总结：
在计算复杂且将在以后的操作中使用的DataFrame上使用localCheckpoint。通常，这些是馈送到unions的节点
在以后要多次使用的DataFrame上使用cache。这通常是一个位于for/while循环之外的DataFrame，该循环将在循环中调用
All Together
您的初始代码
Date_list = [All weeks from: '2021-01-01', to: '2022-01-01'] --> ~50 elements
df_total = spark.createDataframe([], schema)
df_date = []
for date in Date_list:
tmp = df.filter(between [date, date-7days]).withColumn('example', F.lit(date))
........
df2 = df.join(tmp, 'column', 'inner').......
df_date += [df2]
df_total = df_total.unionByName(union_many(*df_date))
return df_total
现在应该看起来像：
# year_marker_df as derived in my code above
year_marker_df = year_marker_df.cache()
df = df.join(year_marker_df, df.my_date_column between year_marker_df.start_date, year_marker_df.end_date)
# Other work previously in your for_loop, resulting in df_total
return df_total
或者，如果你无法重新进行内部循环操作，你可以做一些优化，比如：
Date_list = [All weeks from: '2021-01-01', to: '2022-01-01'] --> ~50 elements
df_total = spark.createDataframe([], schema)
df_date = []
df = df.cache()
for date in Date_list:
tmp = df.filter(between [date, date-7days]).withColumn('example', F.lit(date))
........
df2 = df.join(tmp, 'column', 'inner').......
df2 = df2.localCheckpoint()
df_date += [df2]
df_total = df_total.unionByName(union_many(*df_date))
return df_total