PySpark Pandas UDF写入CSV触发java.io.EOFException求助排查

使用含Pandas UDF的PySpark代码时，执行df.limit(20).collect()及写入20条数据到CSV均正常，但尝试写入100条数据到CSV时触发java.io.EOFException错误。相同逻辑使用普通UDF则无异常，寻求排查方向指导。

基于FlairNLP模型识别命名实体的Pandas UDF #

#Define Flair Function - pandas udf
@pandas_udf("array<string>")
def pu_find_names_flair(iterator: Iterator[pd.Series]) -> Iterator[pd.Series]:
    
   
for textSr in iterator:
    sentences = []
    for text in textSr:
        sentences.append(Sentence(text))
        
    tagger.predict(sentences)
    Flairresult = []
    
    for i, sentence in enumerate(sentences):
        Flairresultentity = []
        # iterate over entities and print each
        for entity in sentence.get_spans('ner'):
            if entity.get_label("ner").value in ["PER", "LOC", "ORG"]:
                Flairresultentity.append((entity.text))
        Flairresult.append(Flairresultentity)        
    yield pd.Series(Flairresult) 

用于查找并替换文本指定字符串的Pandas UDF #

#Mask pii in Call transcripts - pandas udf
@pandas_udf("string")
def pu_mask_all_pii(iterator: Iterator[Tuple[pd.Series, pd.Series]]) -> Iterator[pd.Series]:
"""
Pandas UDF to remove PII from text.
:param: Iterators for two df columns
:return: return iterator
"""

   
for text, pii_list in iterator:

    result = []
    for text, tokens in zip(text, pii_list):
        tokens = sorted(tokens,key=len, reverse=True)
        pat = '|'.join(map(re.escape, tokens))
        text = re.sub(pat, lambda g: 'X' * len(g.group()), str(text), flags=re.IGNORECASE)
        result.append(text)
        
    yield pd.Series(result)

PySpark DataFrame处理代码 #

exclude_pii_words = ['@@@@@@@@','#####']
dfs= dfs.limit(100)
dfs=dfs.withColumn("pii_flairner",pu_find_names_flair(col("CALL_TRANSCRIPT")))
dfs = dfs.withColumn('pii_allmethods', array_except(array_distinct(concat(dfs["pii_bertner"], 
dfs["pii_regex_spacy"], dfs["pii_flairner"])),array(*map(lit,exclude_pii_words))))
dfs = dfs.withColumn('FULL_TRANSCRIPT', pu_mask_all_pii(col("CALL_TRANSCRIPT"), 
                                          col("pii_allmethods")))

dfs = dfs.withColumn('pii_flairner', concat_ws(",",col("pii_flairner")))
dfs = dfs.withColumn('pii_regex_spacy', concat_ws(",",col("pii_regex_spacy")))
dfs = dfs.withColumn('pii_bertner', concat_ws(",",col("pii_bertner")))
dfs = dfs.withColumn('pii_allmethods', concat_ws(",",col("pii_allmethods")))

dfs.limit(20).select('FULL_TRANSCRIPT').collect()  # Works fine

dfs.coalesce(1).write.option("header", "true").mode("overwrite").csv("<filename>") #Errors

异常信息 #

23/03/27 00:44:19 WARN TaskSetManager: Lost task 4.0 in stage 4.0 (TID 14) (10.192.99.70 executor 1): org.apache.spark.SparkException: Python worker exited unexpectedly (crashed)
at org.apache.spark.api.python.BasePythonRunner$ReaderIterator$$anonfun$1.applyOrElse(PythonRunner.scala:678)
at org.apache.spark.api.python.BasePythonRunner$ReaderIterator$$anonfun$1.applyOrElse(PythonRunner.scala:667)
at scala.runtime.AbstractPartialFunction.apply(AbstractPartialFunction.scala:38)
at org.apache.spark.sql.execution.python.PythonArrowOutput$$anon$1.read(PythonArrowOutput.scala:107)
at org.apache.spark.sql.execution.python.PythonArrowOutput$$anon$1.read(PythonArrowOutput.scala:50)
at org.apache.spark.api.python.BasePythonRunner$ReaderIterator.hasNext(PythonRunner.scala:595)
at org.apache.spark.InterruptibleIterator.hasNext(InterruptibleIterator.scala:37)
at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:489)
at scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:458)
at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage2.processNext(Unknown Source)
at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
at org.apache.spark.sql.execution.WholeStageCodegenExec$$anon$1.hasNext(WholeStageCodegenExec.scala:757)
at scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:458)
at org.apache.spark.ContextAwareIterator.hasNext(ContextAwareIterator.scala:39)
at scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:458)
at scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:458)
at scala.collection.Iterator$GroupedIterator.fill(Iterator.scala:1209)
at scala.collection.Iterator$GroupedIterator.hasNext(Iterator.scala:1215)
at scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:458)
at scala.collection.Iterator.foreach(Iterator.scala:941)
at scala.collection.Iterator.foreach$(Iterator.scala:941)
at scala.collection.AbstractIterator.foreach(Iterator.scala:1429)
at org.apache.spark.api.python.PythonRDD$.writeIteratorToStream(PythonRDD.scala:442)
at org.apache.spark.sql.execution.python.PythonUDFRunner$$anon$1.writeIteratorToStream(PythonUDFRunner.scala:53)
at org.apache.spark.api.python.BasePythonRunner$WriterThread.$anonfun$run$1(PythonRunner.scala:521)
at org.apache.spark.util.Utils$.logUncaughtExceptions(Utils.scala:2241)
at org.apache.spark.api.python.BasePythonRunner$WriterThread.run(PythonRunner.scala:313)
Caused by: java.io.EOFException
at java.io.DataInputStream.readInt(DataInputStream.java:392)
at org.apache.spark.sql.execution.python.PythonArrowOutput$$anon$1.read(PythonArrowOutput.scala:88)
... 23 more

依赖版本 #

pyarrow==11.0.0
numpy==1.23.3
pandas==1.4.4

排查方向建议 #

1. Python Worker内存不足：处理100条数据时，Flair模型加载和批量预测可能占用更多内存，导致Worker崩溃。可尝试：

   • 增加Executor内存（--executor-memory）和Python Worker内存配置（spark.python.worker.memory）
   • 调整批处理大小，通过spark.sql.execution.arrow.maxRecordsPerBatch减小单批次数据量，降低内存压力

2. Arrow序列化兼容性问题：EOFException常和Arrow数据传输中断有关，检查：

   • 验证Spark与PyArrow版本兼容性，可尝试降级PyArrow到10.x版本测试
   • 关闭Arrow优化，添加配置spark.sql.execution.arrow.pyspark.enabled=false，改用传统序列化方式验证

3. Flair模型线程安全问题：Pandas UDF的Worker进程可能存在模型共享冲突，将模型初始化移到UDF内部，让每个Worker单独加载：

   @pandas_udf("array<string>")
   def pu_find_names_flair(iterator: Iterator[pd.Series]) -> Iterator[pd.Series]:
       from flair.models import SequenceTagger
       tagger = SequenceTagger.load("ner")  # 移到UDF内部
       for textSr in iterator:
           # 原有逻辑不变
   

4. 异常数据触发崩溃：100条数据中可能存在特殊格式文本，导致Flair预测或正则替换出错。可：

   • 逐步增加数据量（如50条、80条）定位问题数据
   • 在UDF中添加异常捕获，记录错误文本：

     for text in textSr:
         try:
             sentences.append(Sentence(text))
         except Exception as e:
             print(f"Error processing text: {text[:100]} | Exception: {e}")
             sentences.append(Sentence(""))
     

5. Coalesce导致压力集中：coalesce(1)将所有数据集中到一个Executor，加剧内存压力。先去掉该配置直接写入多文件测试，若正常再用repartition(1)替代（repartition会重新洗牌，内存压力相对分散）

内容的提问来源于stack exchange，提问作者Mohan Rayapuvari