我分别通过 pycharm 和 pyspark shell 运行 spark。 我遇到了这个错误:
: java.lang.OutOfMemoryError: Java heap space
at org.apache.spark.api.python.PythonRDD$.readRDDFromFile(PythonRDD.scala:416)
at org.apache.spark.api.python.PythonRDD.readRDDFromFile(PythonRDD.scala)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:231)
at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:381)
at py4j.Gateway.invoke(Gateway.java:259)
at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:133)
at py4j.commands.CallCommand.execute(CallCommand.java:79)
at py4j.GatewayConnection.run(GatewayConnection.java:209)
at java.lang.Thread.run(Thread.java:748)
我的代码是:
from pyspark import SparkContext, SparkConf
from pyspark.sql import HiveContext
import time
if __name__ == '__main__':
print("Started at " + time.strftime("%H:%M:%S"))
conf = (SparkConf()
.setAppName("TestRdd") \
.set('spark.driver.cores', '1') \
.set('spark.executor.cores', '1') \
.set('spark.driver.memory', '16G') \
.set('spark.executor.memory', '9G'))
sc = SparkContext(conf=conf)
rdd = sc.parallelize(range(1000000000),100)
print(rdd.take(10))
print("Finished at " + time.strftime("%H:%M:%S"))
这些是最大内存设置,我可以在集群上设置。我试图将所有内存分配给 1 个核心来创建 rdd。但在我看来,应用程序在分发数据集之前失败了。它无法创建我假设的步骤。我还尝试设置不同数量的分区 100-10000。我已经计算了它需要多少内存,所以 10 亿个 int - 内存中大约 4.5-4.7Gb,比我少,但没有运气。
如何优化并强制运行我的代码?
最佳答案
TL;DR 不要在测试和简单实验之外使用parallelize
。因为您使用 Python 2.7,range
不是惰性的,所以您将实现多种类型的完整值范围:
- 调用后的 Python
list
。 - 稍后将写入磁盘的序列化版本。
- 在 JVM 上加载的序列化副本。
使用 xrange
会有所帮助,但您首先不应使用 parallelize
(或 2018 年的 Python 2)。
如果您想创建一系列值,只需使用 SparkContext.range
range(start, end=None, step=1, numSlices=None)
Create a new RDD of int containing elements from start to end (exclusive), increased by step every element. Can be called the same way as python’s built-in range() function. If called with a single argument, the argument is interpreted as end, and start is set to 0.
所以在你的情况下:
rdd = sc.range(1000000000, numSlices=100)
使用DataFrame
:
from pyspark.sql import SparkSession
spark = SparkSession.builder.getOrCreate()
df = spark.range(1000000000, numPartitions=100)
关于python - PySpark 应用程序失败,出现 java.lang.OutOfMemoryError : Java heap space,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/48708480/