目标是使用自定义项对行进行分类。我在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:超时
1条答案
按热度按时间l5tcr1uw1#
当您可以使用spark内置函数时,请始终避免使用udf。你可以用
when
功能如下:你也可以使用
functools.reduce
动态生成when表达式而不必对所有表达式进行磁带化。例如: