如何将超过 150 亿条数据计算并重塑为 PySpark？答案

【问题标题】：How to calculate and reshape more than 15 bn items of data into PySpark?如何将超过 150 亿条数据计算并重塑为 PySpark？
【发布时间】：2023-02-03 17:27:47
【问题描述】：

我们的用例是从 BQ 读取数据并使用 pandas 和 numpy.reshape 进行计算，将其转换为模型的输入，示例代码如下：

import numpy as np
import pandas as pd

# Source Data
feature = spark.read.format('bigquery') \
    .option('table', TABLE_NAME) \
    .load()

feature_sort = feature.to_pandas_on_spark().sort_values(by = ['col1','col2'], ascending = True).drop(['col1','col3','col5'], axis = 1)

feature_nor = (feature_sort - feature_sort.mean())/(feature_sort.std())

row = int(len(feature_nor)/61)
row2 = 50
col3 = 100

feature_array = np.reshape(feature_nor.values, (row,row2,col3))

feature.to_pandas_on_spark() 会将所有数据收集到驱动程序内存中，对于少量数据它可以工作，但对于超过 150 亿的数据它无法处理。

我尝试将 to_pandas_on_spark() 转换为 spark 数据帧，以便它可以并行计算：

sorted_df = feature.sort('sndr_id').sort('date_index').drop('sndr_id').drop('date_index').drop('cal_dt')
mean_df = sorted_df.select(*[f.mean(c).alias(c) for c in sorted_df.columns])
std_df = sorted_df.select(*[f.stddev(c).alias(c) for c in sorted_df.columns])

由于该功能与pandas api不同，所以我无法验证这些代码并且最后一次重塑操作（np.reshape(feature_nor.values, (row,row2,col3))）数据框不支持此功能，是否有好的解决方案来替换它？

我想知道如何以有效的方式处理 15B 数据而不会内存溢出，包括如何使用 numpy 的 reshape 和 pandas 的计算操作，任何答案都将非常有帮助！

【问题讨论】：

标签： pandas numpy pyspark google-bigquery google-cloud-dataproc

【解决方案1】：

我建议不要在这种大小的数据集上使用 pandas 或 numpy，通常有一些 Spark 函数可以解决您的问题，即使启动 UDF 或在 spark 上使用 pandas 也会带来显着的性能损失。您的重塑标准究竟是什么？也许 pivot 有帮助？

【讨论】：