时间戳

首页 > 解决方案 > 动态构建大型数据框（spark 或 pandas）以导出到 csv 的方法

python - 动态构建大型数据框（spark 或 pandas）以导出到 csv 的方法

问题描述

我有一个使用 spark.read 导入数据块的 csv。这个大文件包含每天的记录/交易。我将数据框缩减为 5 列，并保持 500,000 行不变。我正在尝试构建此源文件的汇总表，以一个月级别（汇总）表示这些记录/交易。

该脚本有一个 filter/groupby/sum 命令，它返回一行，将数据汇总为一个月的计数。查询返回的行如下所示：

+---------+---------+-------+-------------+
|  Country|StockCode|YYYY-MM|sum(Quantity)|
+---------+---------+-------+-------------+
|Singapore|        M| 2011-4|           10|
+---------+---------+-------+-------------+

该脚本遍历源数据帧并每次返回。我无法使用此脚本的输出（显示或 csv 导出）。在 pyspark 和 pandas 中我都有问题。我不确定如何堆叠查询结果以及它应该采用什么形式？

#Pandas 如果我在 pandas 中执行此操作，脚本生成文件需要很长时间（我相信 pandas + 我这样做效率不高导致持续时间延长）~ 2.5 小时。display 和 write.csv 命令的运行速度相当快，并且在大约几秒钟内完成。

#Pyspark 如果我在 pyspark 中执行此操作，脚本大约需要 10 分钟才能完成，但显示和导出会崩溃。笔记本要么返回超时错误，要么重新启动，要么引发崩溃错误。

该方法是否应该是动态创建列表列表，并在完全构建后将其转换为数据框以供使用？我一直在尝试我遇到的所有方法，但我似乎没有取得任何进展。

这是生成结果的代码

#officeSummaryDFBefore
column_names = "Country|StockCode|YYYY-MM|Quantity"
monthlyCountsBeforeImpactDate = spark.createDataFrame(
  [
    tuple('' for i in column_names.split("|"))
  ],
  column_names.split("|")
).where("1=0")

monthlyCountsBeforeImpacteDateRow = spark.createDataFrame(
  [
    tuple('' for i in column_names.split("|"))
  ],
  column_names.split("|")
).where("1=0")

try :
  for country in country_lookup :
    country = country[0]
    print(country_count, " country(s) left")
    country_count = country_count - 1
    for stockCode in stockCode_lookup :
      stockCode = stockCode[0]
      monthlyCountsBeforeImpacteDateRow = dataBeforeImpactDate.filter((col("Country").rlike(country)) & (col("StockCode").rlike(stockCode))).groupby("Country", "StockCode", "YYYY-MM").sum()
      monthlyCountsBeforeImpacteDateRow.show()
      dfsCountsBefore = [monthlyCountsBeforeImpacteDateRow, monthlyCountsBeforeImpactDate]
      monthlyCountsBeforeImpactDate = reduce(DataFrame.union, dfsCountsBefore)
      
except Exception as e:
  print(e)

我在循环内声明 dfsCountsBeforeImpactDate 这似乎不正确，但是当它在外部时它返回为 NULL。

标签： pythonpandasapache-sparkpysparkdatabricks

解决方案

IIUC 您正在查找国家和股票以限制行，然后对它们进行分组以生成聚合。

为什么不完全过滤df然后分组

df = dataBeforeImpactDate

df = df.filter(col('country').isin(country_lookup) & col('stock').isin(stock_lookup))

df = df.groupby("Country", "StockCode", "YYYY-MM").sum()

df.show()

这会更快，因为您不需要循环过滤器，也不需要联合。

推荐阅读