在使用RDD时出现Py4JJavaError Pyspark错误

Question

我遇到了以下错误：

py4j.protocol.Py4JJavaError: An error occurred while calling z:org.apache.spark.api.python.PythonRDD.collectAndServe.

在这一行：

result = df.select('student_age').rdd.flatMap(lambda x: x).collect()

“student_age” 是一个列名。它一直运行良好，直到上周出现了这个错误。

请问有人对此有任何见解吗？

- Slickmind

---

你能分享更多关于错误日志和你的代码的信息吗？ - Jonathan Lam

这是完整的堆栈跟踪吗？另外，“上周还运行良好”——您最近更新了任何内容，例如Java版本？此外，您在哪里运行这个程序？ - viggnah

没有任何变化 - Slickmind

所有日志： https://ctxt.io/2/AADge2-UFg - Slickmind

@Slickmind，你能展示一下data_percentage和count_percentage函数的代码吗？ - Jonathan Lam

Answer 1

使用collect是很危险的，因为它容易导致内存不足错误。建议删除它。您也不需要使用rdd，可以使用数据框来完成此操作：

result = df.select(explode(df['student_age'])) #returns a dataFrame
#write code to use a data frame instead of any array.

如果没有其他变化，很可能数据发生了变化，并且最终超出了内存大小。

也有可能您有新的“坏”数据导致错误。

无论哪种情况，您都可以通过查找此(OOM)来证明这一点，或通过打印数据来证明数据有问题。

def f(row):
    print(row.student_age)

result.foreach(f) # used for simple stuff that doesn't require heavy initialization.

如果这个方法有效，你可能想要将你的代码拆分成foreachPartition使用。这样可以让你在每个执行器的内存中对每个值进行数学计算。唯一的问题是，在下面的fun中，由于你正在执行此代码，因此不能引用任何使用sparkContext的内容。（仅限Python代码，而不是Pyspark）。

def f(rows):
    #intialize a database connection here
    for row in rows:
        print(row.student_age) # do stuff with student_age
    #close database connection here

result.foreachPartition(f) # used for things that need heavy initialization

Spark foreachPartition与foreach | 该使用哪个？

- Matt Andruff

---

谢谢Matt！让我试试看！ - Slickmind

Answer 2

这个问题已经解决了，以下是答案：

result = [i[0] for i in df.select('student_age').toLocalIterator()]

- Slickmind

---

这可能适用于您的用例，因为使用迭代器而不是collect()可以减少内存压力。此时您并没有在做pyspark，而是在进行Python编码。也就是说，原则上，您正在使用“小”数据方法而不是使用“大”数据方法。我鼓励您考虑更大的数据，并使用dataFrames而不是使用Python数组。这样会更具可扩展性。 - Matt Andruff

我遇到了这个错误：

---

File "C:\Users\Jarvis\AppData\Local\Programs\Python\Python38\lib\site-packages\pyspark\sql\column.py", line 470, in __iter__ raise TypeError("Column is not iterable")

---

结果 = df.select(explode(df['student_age'])) unique_result = list(set([j for i in result for j in xs])) - Slickmind

你仍在使用Python而不是pyspark。要使列唯一，只需在其上调用“distinct()”。更多地学习pyspark可能会对你有所裨益。显然你懂Python。 - Matt Andruff