英文:
TypeError in pySpark UDF functions
问题
我有这个函数:
def ead(lista):
ind_mmff, isdebala, isfubala, k1, k2, ead = lista
try:
isdebala = float(isdebala)
isfubala = float(isfubala)
k1 = float(k1)
k2 = float(k2)
ead = float(ead)
except ValueError:
return '错误:无效的输入'
min_deb = min(0, isdebala)
min_fub = min(0, isfubala)
if ind_mmff == '0':
ead_dai = abs(min_deb * k1 / 100 + min_fub * k2 / 100)
else:
ead_dai = ead
return ead_dai
然后,我定义一个用户定义函数(UDF),如下所示:
ead_udf = udf(lambda z: ead(z), FloatType())
目标是在我的数据框(df)中创建一个名为ead_calc
的列,如下所示:
df = df.withColumn('ead_calc', ead_udf(array(df.ind_mmff, df.isdebala, df.isfubala, df.k1, df.k2, df.ead_final_motor)))
执行df.select('ead_calc').show()
后,会引发以下错误:
Py4JJavaError: An error occurred while calling o3026.showString.
: org.apache.spark.SparkException: Job aborted due to stage failure: Task 3 in stage 813.0 failed 4 times, most recent failure: Lost task 3.3 in stage 813.0 (TID 12054, csslncclowp0006.unix.aacc.corp, executor 2): org.apache.spark.api.python.PythonException: Traceback (most recent call last):
File "/opt/cloudera/parcels/SPARK2-2.4.0.cloudera2-1.cdh5.13.3.p0.1041012/lib/spark2/python/lib/pyspark.zip/pyspark/worker.py", line 377, in main
process()
File "/opt/cloudera/parcels/SPARK2-2.4.0.cloudera2-1.cdh5.13.3.p0.1041012/lib/spark2/python/lib/pyspark.zip/pyspark/worker.py", line 372, in process
serializer.dump_stream(func(split_index, iterator), outfile)
File "/opt/cloudera/parcels/SPARK2-2.4.0.cloudera2-1.cdh5.13.3.p0.1041012/lib/spark2/python/lib/pyspark.zip/pyspark/serializers.py", line 345, in dump_stream
self.serializer.dump_stream(self._batched(iterator), stream)
File "/opt/cloudera/parcels/SPARK2-2.4.0.cloudera2-1.cdh5.13.3.p0.1041012/lib/spark2/python/lib/pyspark.zip/pyspark/serializers.py", line 141, in dump_stream
for obj in iterator:
File "/opt/cloudera/parcels/SPARK2-2.4.0.cloudera2-1.cdh5.13.3.p0.1041012/lib/spark2/python/lib/pyspark.zip/pyspark/serializers.py", line 334, in _batched
for item in iterator:
File "<string>", line 1, in <lambda>
File "/opt/cloudera/parcels/SPARK2-2.4.0.cloudera2-1.cdh5.13.3.p0.1041012/lib/spark2/python/lib/pyspark.zip/pyspark/worker.py", line 85, in <lambda>
return lambda *a: f(*a)
File "/opt/cloudera/parcels/SPARK2-2.4.0.cloudera2-1.cdh5.13.3.p0.1041012/lib/spark2/python/lib/pyspark.zip/pyspark/util.py", line 99, in wrapper
return f(*args, **kwargs)
File "<ipython-input-93-25e605cffdae>", line 1, in <lambda>
File "<ipython-input-92-a1937fe32209>", line 12, in ead
TypeError: _() takes 1 positional argument but 2 were given
错误位于min_deb = min(0, isdebala)
。不知道如何解决这个问题,因为min
函数显然需要两个参数。
目标是在我的数据框(df)中创建一个名为ead_calc
的列,如下所示:
df = df.withColumn('ead_calc', ead_udf(array(df.ind_mmff, df.isdebala, df.isfubala, df.k1, df.k2, df.ead_final_motor)))
英文:
I've got this function:
def ead(lista):
ind_mmff, isdebala, isfubala, k1, k2, ead = lista
try:
isdebala = float(isdebala)
isfubala = float(isfubala)
k1 = float(k1)
k2 = float(k2)
ead = float(ead)
except ValueError:
return 'Error: invalid input'
min_deb = min(0, isdebala)
min_fub = min(0, isfubala)
if ind_mmff == '0':
ead_dai = abs(min_deb * k1 / 100 + min_fub * k2 / 100)
else:
ead_dai = ead
return ead_dai
Afterwards, I define a UDF such as:
ead_udf = udf(lambda z: ead(z), FloatType())
The aim is to create a ead_calc
column in my df dataframe such as:
df = df.withColumn('ead_calc', ead_udf (array(df.ind_mmff, df.isdebala, df.isfubala, df.k1, df.k2, df.ead_final_motor)))
After executing df.select('ead_calc').show()
the following error raises:
Py4JJavaError: An error occurred while calling o3026.showString.
: org.apache.spark.SparkException: Job aborted due to stage failure: Task 3 in stage 813.0 failed 4 times, most recent failure: Lost task 3.3 in stage 813.0 (TID 12054, csslncclowp0006.unix.aacc.corp, executor 2): org.apache.spark.api.python.PythonException: Traceback (most recent call last):
File "/opt/cloudera/parcels/SPARK2-2.4.0.cloudera2-1.cdh5.13.3.p0.1041012/lib/spark2/python/lib/pyspark.zip/pyspark/worker.py", line 377, in main
process()
File "/opt/cloudera/parcels/SPARK2-2.4.0.cloudera2-1.cdh5.13.3.p0.1041012/lib/spark2/python/lib/pyspark.zip/pyspark/worker.py", line 372, in process
serializer.dump_stream(func(split_index, iterator), outfile)
File "/opt/cloudera/parcels/SPARK2-2.4.0.cloudera2-1.cdh5.13.3.p0.1041012/lib/spark2/python/lib/pyspark.zip/pyspark/serializers.py", line 345, in dump_stream
self.serializer.dump_stream(self._batched(iterator), stream)
File "/opt/cloudera/parcels/SPARK2-2.4.0.cloudera2-1.cdh5.13.3.p0.1041012/lib/spark2/python/lib/pyspark.zip/pyspark/serializers.py", line 141, in dump_stream
for obj in iterator:
File "/opt/cloudera/parcels/SPARK2-2.4.0.cloudera2-1.cdh5.13.3.p0.1041012/lib/spark2/python/lib/pyspark.zip/pyspark/serializers.py", line 334, in _batched
for item in iterator:
File "<string>", line 1, in <lambda>
File "/opt/cloudera/parcels/SPARK2-2.4.0.cloudera2-1.cdh5.13.3.p0.1041012/lib/spark2/python/lib/pyspark.zip/pyspark/worker.py", line 85, in <lambda>
return lambda *a: f(*a)
File "/opt/cloudera/parcels/SPARK2-2.4.0.cloudera2-1.cdh5.13.3.p0.1041012/lib/spark2/python/lib/pyspark.zip/pyspark/util.py", line 99, in wrapper
return f(*args, **kwargs)
File "<ipython-input-93-25e605cffdae>", line 1, in <lambda>
File "<ipython-input-92-a1937fe32209>", line 12, in ead
TypeError: _() takes 1 positional argument but 2 were given
The error is located at min_deb = min(0, isdebala)
. Don't know how to solve this issue since min function obviously requires 2 arguments.
The aim is to create a ead_calc column in my df dataframe such as:
df = df.withColumn('ead_calc', ead_udf (array(df.ind_mmff, df.isdebala, df.isfubala, df.k1, df.k2, df.ead_final_motor)))
答案1
得分: 1
I think you have imported the wrong min
function, I guess you have imported the one from pyspark by using from pyspark.sql.functions import *
, the pyspark min function takes only one argument (column) but the python one takes two arguments
Trying to import only the needed functions and it seems working (Just added some random input)
from pyspark.sql.functions import udf, array
from pyspark.sql.types import StructField, StructType, FloatType
def ead(lista):
ind_mmff, isdebala, isfubala, k1, k2, ead = lista
try:
isdebala = float(isdebala)
isfubala = float(isfubala)
k1 = float(k1)
k2 = float(k2)
ead = float(ead)
except ValueError:
return 'Error: invalid input'
min_deb = min(0, isdebala)
min_fub = min(0, isfubala)
if ind_mmff == '0':
ead_dai = abs(min_deb * k1 / 100 + min_fub * k2 / 100)
else:
ead_dai = ead
return ead_dai
ead_udf = udf(lambda z: ead(z), FloatType())
schema = StructType([
StructField('ind_mmff', FloatType(), True),
StructField('isdebala', FloatType(), True),
StructField('isfubala', FloatType(), True),
StructField('k1', FloatType(), True),
StructField('k2', FloatType(), True),
StructField('ead_final_motor', FloatType(), True)
])
df = spark.createDataFrame(data=[(1.0, 2.0, 3.0, 4.0, 5.0, 6.0)],schema=schema)
df = df.withColumn('ead_calc', ead_udf (array(df.ind_mmff, df.isdebala, df.isfubala, df.k1, df.k2, df.ead_final_motor)))
df.show()
+--------+--------+--------+---+---+---------------+--------+
|ind_mmff|isdebala|isfubala| k1| k2|ead_final_motor|ead_calc|
+--------+--------+--------+---+---+---------------+--------+
| 1.0| 2.0| 3.0|4.0|5.0| 6.0| 6.0|
+--------+--------+--------+---+---+---------------+--------+
英文:
I think you have imported the wrong min
function, I guess you have imported the one from pyspark by using from pyspark.sql.functions import *
, the pyspark min function takes only one argument (column) but the python one takes two arguments
Trying to import only the needed functions and it seems working (Just added some random input)
from pyspark.sql.functions import udf, array
from pyspark.sql.types import StructField, StructType, FloatType
def ead(lista):
ind_mmff, isdebala, isfubala, k1, k2, ead = lista
try:
isdebala = float(isdebala)
isfubala = float(isfubala)
k1 = float(k1)
k2 = float(k2)
ead = float(ead)
except ValueError:
return 'Error: invalid input'
min_deb = min(0, isdebala)
min_fub = min(0, isfubala)
if ind_mmff == '0':
ead_dai = abs(min_deb * k1 / 100 + min_fub * k2 / 100)
else:
ead_dai = ead
return ead_dai
ead_udf = udf(lambda z: ead(z), FloatType())
schema = StructType([
StructField('ind_mmff', FloatType(), True),
StructField('isdebala', FloatType(), True),
StructField('isfubala', FloatType(), True),
StructField('k1', FloatType(), True),
StructField('k2', FloatType(), True),
StructField('ead_final_motor', FloatType(), True)
])
df = spark.createDataFrame(data=[(1.0, 2.0, 3.0, 4.0, 5.0, 6.0)],schema=schema)
df = df.withColumn('ead_calc', ead_udf (array(df.ind_mmff, df.isdebala, df.isfubala, df.k1, df.k2, df.ead_final_motor)))
df.show()
+--------+--------+--------+---+---+---------------+--------+
|ind_mmff|isdebala|isfubala| k1| k2|ead_final_motor|ead_calc|
+--------+--------+--------+---+---+---------------+--------+
| 1.0| 2.0| 3.0|4.0|5.0| 6.0| 6.0|
+--------+--------+--------+---+---+---------------+--------+
通过集体智慧和协作来改善编程学习和解决问题的方式。致力于成为全球开发者共同参与的知识库,让每个人都能够通过互相帮助和分享经验来进步。
评论