TypeError in pySpark UDF functions

huangapple go评论58阅读模式
英文:

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 &#39;Error: invalid input&#39;        
    min_deb = min(0, isdebala)
    min_fub = min(0, isfubala)
    
    if ind_mmff == &#39;0&#39;:
        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(&#39;ead_calc&#39;, ead_udf (array(df.ind_mmff, df.isdebala, df.isfubala, df.k1, df.k2, df.ead_final_motor)))

After executing df.select(&#39;ead_calc&#39;).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 &quot;/opt/cloudera/parcels/SPARK2-2.4.0.cloudera2-1.cdh5.13.3.p0.1041012/lib/spark2/python/lib/pyspark.zip/pyspark/worker.py&quot;, line 377, in main
    process()
  File &quot;/opt/cloudera/parcels/SPARK2-2.4.0.cloudera2-1.cdh5.13.3.p0.1041012/lib/spark2/python/lib/pyspark.zip/pyspark/worker.py&quot;, line 372, in process
    serializer.dump_stream(func(split_index, iterator), outfile)
  File &quot;/opt/cloudera/parcels/SPARK2-2.4.0.cloudera2-1.cdh5.13.3.p0.1041012/lib/spark2/python/lib/pyspark.zip/pyspark/serializers.py&quot;, line 345, in dump_stream
    self.serializer.dump_stream(self._batched(iterator), stream)
  File &quot;/opt/cloudera/parcels/SPARK2-2.4.0.cloudera2-1.cdh5.13.3.p0.1041012/lib/spark2/python/lib/pyspark.zip/pyspark/serializers.py&quot;, line 141, in dump_stream
    for obj in iterator:
  File &quot;/opt/cloudera/parcels/SPARK2-2.4.0.cloudera2-1.cdh5.13.3.p0.1041012/lib/spark2/python/lib/pyspark.zip/pyspark/serializers.py&quot;, line 334, in _batched
    for item in iterator:
  File &quot;&lt;string&gt;&quot;, line 1, in &lt;lambda&gt;
  File &quot;/opt/cloudera/parcels/SPARK2-2.4.0.cloudera2-1.cdh5.13.3.p0.1041012/lib/spark2/python/lib/pyspark.zip/pyspark/worker.py&quot;, line 85, in &lt;lambda&gt;
    return lambda *a: f(*a)
  File &quot;/opt/cloudera/parcels/SPARK2-2.4.0.cloudera2-1.cdh5.13.3.p0.1041012/lib/spark2/python/lib/pyspark.zip/pyspark/util.py&quot;, line 99, in wrapper
    return f(*args, **kwargs)
  File &quot;&lt;ipython-input-93-25e605cffdae&gt;&quot;, line 1, in &lt;lambda&gt;
  File &quot;&lt;ipython-input-92-a1937fe32209&gt;&quot;, 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(&#39;ead_calc&#39;, 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 &#39;Error: invalid input&#39;        
    min_deb = min(0, isdebala)
    min_fub = min(0, isfubala)
    
    if ind_mmff == &#39;0&#39;:
        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(&#39;ind_mmff&#39;, FloatType(), True),
  StructField(&#39;isdebala&#39;, FloatType(), True),
  StructField(&#39;isfubala&#39;, FloatType(), True),
  StructField(&#39;k1&#39;, FloatType(), True),
  StructField(&#39;k2&#39;, FloatType(), True),
  StructField(&#39;ead_final_motor&#39;, FloatType(), True)
  ])

df = spark.createDataFrame(data=[(1.0, 2.0, 3.0, 4.0, 5.0, 6.0)],schema=schema)

df = df.withColumn(&#39;ead_calc&#39;, 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|
+--------+--------+--------+---+---+---------------+--------+

huangapple
  • 本文由 发表于 2023年6月6日 01:13:50
  • 转载请务必保留本文链接:https://go.coder-hub.com/76408657.html
匿名

发表评论

匿名网友

:?: :razz: :sad: :evil: :!: :smile: :oops: :grin: :eek: :shock: :???: :cool: :lol: :mad: :twisted: :roll: :wink: :idea: :arrow: :neutral: :cry: :mrgreen:

确定