无法从spark dataframe导出数据

Question

作为测试，我使用spark NLP解析了50万条tweet。数据帧看起来没问题。我将数组转换为字符串。使用

from pyspark.sql.functions import udf
from pyspark.sql.types import StringType

def array_to_string(my_list):
    return '[' + ','.join([str(elem) for elem in my_list]) + ']'

array_to_string_udf = udf(array_to_string, StringType())

result = result.withColumn('token', array_to_string_udf(result["token"])).withColumn('ner', array_to_string_udf(result["ner"])).withColumn('embeddings', array_to_string_udf(result["embeddings"])).withColumn('ner_chunk', array_to_string_udf(result["ner_chunk"])).withColumn('document', array_to_string_udf(result["document"]))

数据帧看起来没问题。但是，每当我尝试将其转换为pandas，将其导出为csv时，我都会收到以下错误

PythonException: 
  An exception was thrown from the Python worker. Please see the stack trace below.
Traceback (most recent call last):
  File "C:\spark\spark-3.1.2-bin-hadoop3.2\python\lib\pyspark.zip\pyspark\worker.py", line 584, in main
  File "C:\spark\spark-3.1.2-bin-hadoop3.2\python\lib\pyspark.zip\pyspark\serializers.py", line 562, in read_int
    length = stream.read(4)
  File "C:\ProgramData\Anaconda3\lib\socket.py", line 669, in readinto
    return self._sock.recv_into(b)
socket.timeout: timed out

这让我觉得斯帕克不是在和蟒蛇说话。有人知道问题出在哪里吗？

Answer 1

当调用toPandas时

将所有数据加载到驱动程序的内存中

调用toPandas的效果与调用collect的效果基本相同。

将数据帧的内容写入csv的更好的方法是直接使用PySpark的DataFrameWriter：

result.write.csv('my_result.csv')

编辑:可能与问题没有直接关系，但可以用原生Spark函数(lit、concat和concat_ws)替换自定义函数：

from pyspark.sql import functions as F

result.withColumn("token", F.concat(F.lit("["), F.concat_ws(",", "token"),F.lit("]")))....

用Spark函数替换udf将提高性能。