无法从 spark 数据框中导出数据

如何解决无法从 spark 数据框中导出数据

我使用 spark NLP 将 50 万条推文解析为测试。数据框看起来不错。我将数组转换为字符串。使用

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

这让我觉得 spark 不是在和 python 对话。有谁知道可能是什么问题？

解决方法

调用 toPandas 时

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

调用toPandas的效果和调用collect的效果基本一样。

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

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

编辑：可能与问题没有直接关系，但可以用原生 Spark 函数（lit、concat 和 {{3} }):

from pyspark.sql import functions as F

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

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