【问题标题】:Why does pySpark crash when using Apache Arrow for string types?为什么在将 Apache Arrow 用于字符串类型时 pySpark 会崩溃?
【发布时间】:2021-08-05 22:18:15
【问题描述】:

为了在大型数据集上获得一些异常值图,我需要将 spark DataFrame 转换为 pandas。 Turing to Apache Arrow 一个简单的运行在将 x 转换为字符串时使我的 pyspark 控制台崩溃(没有转换它可以正常工作),为什么?

Using Python version 3.8.9 (default, Apr 10 2021 15:47:22)
Spark context Web UI available at http://6d0b1018a45a:4040
Spark context available as 'sc' (master = local[*], app id = local-1621164597906).
SparkSession available as 'spark'.
>>> import time
>>> from pyspark.sql.functions import rand
>>> from pyspark.sql import functions as F
>>> spark = SparkSession.builder.appName("Console_Test").getOrCreate()
>>> spark.conf.set("spark.sql.execution.arrow.enabled", "true")
21/05/16 11:31:03 WARN SQLConf: The SQL config 'spark.sql.execution.arrow.enabled' has been deprecated in Spark v3.0 and may be removed in the future. Use 'spark.sql.execution.arrow.pyspark.enabled' instead of it.
>>> a_df = spark.range(1 << 25).toDF("id").withColumn("x", rand())
>>> a_df = a_df.withColumn("id", F.col("id").cast("string"))
>>> start_t = time.time()
>>> a_pd = a_df.toPandas()
Killed                                                                          
#

此外,我注意到 spark.conf.set("spark.sql.execution.arrow.maxRecordsPerBatch", "5000") 之类的选项似乎没有效果,因为 Web ui 显示分配给任务的记录明显超过 5000。

任何关于如何解决 pyspark 控制台崩溃或更直接渲染大散点图的指示将不胜感激 - 我(未成功)试图找到一种方法来应用 Table.to_pandas(split_blocks=True, self_destruct=True),但没有从火花中获得有效的结构DataFrame.

【问题讨论】:

  • 这个来自the docs的sn-p可能是相关的,Note that even with Arrow, DataFrame.toPandas() results in the collection of all records in the DataFrame to the driver program and should be done on a small subset of the data.
  • 您应该能够使用pandas_udf 来减少内存使用量。这就是 maxRecordsPerBatch 属性的用途。
  • @Pace 谢谢。我不知道该限制仅适用于 pandas_udf。这解释了观察到的行为。

标签: dataframe pyspark pyarrow apache-arrow


【解决方案1】:

您尝试将 33.5 mio (2^25) 行转换为 Pandas 数据框。这将导致 OutOfMemoryError,因为所有数据都将传输到 Spark 驱动程序。

查找异常值的一种方法是计算列xhistogram,然后在创建Pandas 数据框之前将a_df 过滤到Spark 中的相关bin:

hist = a_df.select("x").rdd.flatMap(lambda x: x).histogram(10) #create 10 bins

hist 是两个数组的元组:第一个数组包含 bin 的边界,第二个数组包含每个 bin 中的元素数:

([1.7855041778425118e-08,
  0.1000000152099446,
  0.20000001256484742,
  0.30000000991975023,
  0.40000000727465307,
  0.5000000046295558,
  0.6000000019844587,
  0.6999999993393615,
  0.7999999966942644,
  0.8999999940491672,
  0.99999999140407],
 [3355812,
  3356891,
  3352364,
  3352438,
  3357564,
  3356213,
  3354933,
  3355144,
  3357241,
  3355832])

rand 创建均匀分布的随机数,因此这种情况下的直方图不是很有趣。但对于现实世界的分布,直方图会很有用。

【讨论】:

  • 谢谢。我试图以某种方式避免在驱动程序处收集,同时保持 pandas 的高级功能——3340 万只是完整数据的 1% 样本。也许在未来的某个时候,可能会有一种结构,其行为类似于用户的 pandas 数据框并利用 Spark 的规模。同时我会满足于火花:)
猜你喜欢
  • 1970-01-01
  • 2016-09-16
  • 2014-06-15
  • 2014-04-30
  • 2018-02-15
  • 1970-01-01
  • 2011-04-17
  • 1970-01-01
  • 2021-11-19
相关资源
最近更新 更多