pyspark，错误为self.\u sock.recv\u into(b)socket.timeout:超时

62lalag4 于 2021-07-13 发布在 Spark

关注(0)|答案(1)|浏览(511)

目标是使用自定义项对行进行分类。我在windows上使用pyspark。
使用简单的函数或操作（如filter）似乎是可行的。
任何关于如何解决超时/套接字故障的指导都会很有帮助（请参阅下面的错误）。
数据中没有空值。

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\serializers.py”，第593行，read\int length=stream.read（4）文件“c:\programdata\anaconda3\lib\socket.py”，第669行，in readinto return self.\u sock.recv\u into（b）socket.timeout:超时

python apache-spark pyspark apache-spark-sql pyspark-dataframes

来源：https://stackoverflow.com/questions/66071150/pyspark-with-error-self-sock-recv-intob-socket-timeout-timed-out

1条答案

按热度按时间

l5tcr1uw1#

当您可以使用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)

赞(0）回复(0）举报 2021-07-13

我来回答

pyspark，错误为self.\u sock.recv\u into(b)socket.timeout:超时

1条答案

相关问题

热门标签

最新问答