Spark可以使用一个小收藏

Question

我对Spark，其驱动程序和OOM问题有问题。

目前，我有一个数据框架，该数据框架正在使用多个加入的来源（实际上是镶木式格式的不同表），并且有成千上万个单元。他们的日期代表了记录的创建日期，显然是其中的日期。

我做以下操作：

from pyspark.sql.functions import year, month

# ...

selectionRows = inputDataframe.select(year('registration_date').alias('year'), month('registration_date').alias('month')).distinct()
selectionRows.show() # correctly shows 8 tuples
selectionRows = selectionRows.collect() # goes heap space OoM
print(selectionRows)

阅读内存消耗统计信息表明，驱动程序不超过约60％。我认为驱动程序应仅加载独特的子集，而不是整个数据框架。

我想念什么吗？是否可以以更聪明的方式收集这几行？我需要它们作为推高谓词来加载辅助数据框架。

非常感谢！

strong

阅读评论并详细说明我的个人需求后，我在每个“加入/详细”步骤中都捕获了数据

• 编辑/解决方案< /
• 框
• 应用缓存转换
• 打印计数以跟踪基数（主要是用于跟踪 /调试目的），因此应用所有转换 +缓存
• 不受欢迎的先前的浮雕步骤的缓存（如果可用）（tick / tock paradigm），

这降低了一些复杂的ETL作业降至原始时间的20％（如前所述，它在每个计数上都应用了上一步的转换）。

经验教训:)

Answer 1

阅读评论后，我详细介绍了用例。

如问题所述，我在“目标数据框架”中连接了几个表，在每次迭代中，我都会进行一些转换，例如：

# n-th table work
target = target.join(other, how='left')
target = target.filter(...)
target = target.withColumn('a', 'b')
target = target.select(...)
print(f'Target after table "other": {target.count()}')

慢速 / OOM的问题是，Spark被迫从由于结束<代码>计数< / code>，在每个表上开始完成，在每个表 /迭代时都会慢慢慢。

我发现的解决方案是在每次迭代中缓存数据框，就像这样：

cache: DataFrame = null

# ...

# n-th table work
target = target.join(other, how='left')
target = target.filter(...)
target = target.withColumn('a', 'b')
target = target.select(...)

target = target.cache()
target_count = target.count() # actually do the cache
if cache:
  cache.unpersist() # free the memory from the old cache version
cache = target

print(f'Target after table "other": {target_count}')