Pyspark 错误 self._sock.recv_into(b) socket.timeout: 超时答案

【问题标题】：Pyspark with error self._sock.recv_into(b) socket.timeout: timed outPyspark 错误 self._sock.recv_into(b) socket.timeout: 超时
【发布时间】：2021-05-10 06:14:33
【问题描述】：

目标是使用 UDF 对行进行分类。我在 Windows 上使用 pyspark。

使用像过滤器这样的简单函数或操作似乎可以工作。

有关如何解决超时/套接字故障的任何指导都会有所帮助（请参阅下面的错误）。

数据中没有空值。

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

def BreakDown(arr_value):
    start_year = arr_value[0]
    start_month = arr_value[1]
    end_year = arr_value[2]
    end_month = arr_value[3]
    curr_year = arr_value[4]
    curr_month = arr_value[5]
    if   (curr_year == start_year) & (curr_month >= start_month) : return 1
    elif   (curr_year == end_year) & (curr_month <= end_month) : return 1
    elif   (curr_year > start_year) & (curr_year < end_year) : return 1
    else: return 0

    
udfBreakDown = udf(BreakDown, IntegerType())

temp = temp.withColumn('include', udfBreakDown(F.struct('start_year','start_month','end_year','end_month','curr_year','curr_month')))

PythonException：Python 工作者抛出异常。请参阅下面的堆栈跟踪。回溯（最近一次通话最后一次）：
文件 "E:\spark\spark-3.0.1-bin-hadoop2.7\python\lib\pyspark.zip\pyspark\worker.py", 第 585 行，在主文件中 "E:\spark\spark-3.0.1-bin-hadoop2.7\python\lib\pyspark.zip\pyspark\serializers.py", 第 593 行，在 read_int 中长度 = stream.read(4) 文件“C:\ProgramData\Anaconda3\lib\socket.py”，第 669 行，在 readinto return self._sock.recv_into(b) socket.timeout: 超时

【问题讨论】：

这里不需要UDF，使用when函数。
它可以在 linux 机器上运行。而且我宁愿写一个可读的 udf - 特别是随着条件的增加。
UDF 在 pyspark 中以性能不佳而闻名。请不要使用它们，除非您想要实现的目标没有 spark 内置函数。为了便于阅读，您还可以编写返回 when 列表达式的包装函数。
好的 - 让我按照你的建议试试。
Sir BlackBishop - 可以确认何时工作，没有故障代替 udf。请包括作为我接受的答案。不过，如果我再次遇到这种情况，通常是否可以配置 wiat 时间？

标签： python apache-spark pyspark apache-spark-sql pyspark-dataframes

【解决方案1】：

当您可以使用 Spark 内置函数时，请始终避免使用 UDF。您可以使用when 函数重写您的逻辑，如下所示：

from pyspark.sql import functions as F

def get_include_col():
    c = F.when((F.col("curr_year") == F.col("start_year")) & (F.col("curr_month") >= F.col("start_month")), F.lit(1)) \
        .when((F.col("curr_year") == F.col("end_year")) & (F.col("curr_month") <= F.col("end_month")), F.lit(1)) \
        .when((F.col("curr_year") > F.col("start_year")) & (F.col("curr_year") < F.col("end_year")), F.lit(1)) \
        .otherwise(F.lit(0))
    return c


temp = temp.withColumn('include', get_include_col())

您还可以使用functools.reduce 来动态生成 when 表达式，而无需将它们全部记录下来。例如：

import functools
from pyspark.sql import functions as F

cases = [
    ("curr_year = start_year and curr_month >= start_month", 1),
    ("curr_year = end_year and curr_month <= end_month", 1),
    ("curr_year > start_year and curr_year < end_year", 1)
]

include_col = functools.reduce(
    lambda acc, x: acc.when(F.expr(x[0]), F.lit(x[1])),
    cases,
    F
).otherwise(F.lit(0))

temp = temp.withColumn('include', include_col)

【讨论】：