快捷導(dǎo)航

pyspark自定義UDAF函數(shù)調(diào)用報錯問題解決

更新時間：2022年06月08日 16:16:36 作者：未入坑的小白

這篇文章主要為大家介紹了pyspark自定義UDAF函數(shù)調(diào)用報錯問題解決，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進步，早日升職加薪

問題場景：

在SparkSQL中，因為需要用到自定義的UDAF函數(shù)，所以用pyspark自定義了一個，但是遇到了一個問題，就是自定義的UDAF函數(shù)一直報

AttributeError: 'NoneType' object has no attribute '_jvm'

在此將解決過程記錄下來

問題描述

在新建的py文件中，先自定義了一個UDAF函數(shù)，然后在 if __name__ == '__main__': 中調(diào)用，死活跑不起來，一遍又一遍的對源碼，看起來自定義的函數(shù)也沒錯：過程如下:

import decimal
import os
import pandas as pd
from pyspark.sql import SparkSession
from pyspark.sql import functions as F
os.environ['SPARK_HOME'] = '/export/server/spark'
os.environ["PYSPARK_PYTHON"] = "/root/anaconda3/bin/python"
os.environ["PYSPARK_DRIVER_PYTHON"] = "/root/anaconda3/bin/python"
@F.pandas_udf('decimal(17,12)')
def udaf_lx(qx: pd.Series, lx: pd.Series) -> decimal:
    # 初始值 也一定是decimal類型
    tmp_qx = decimal.Decimal(0)
    tmp_lx = decimal.Decimal(0)
    for index in range(0, qx.size):
        if index == 0:
            tmp_qx = decimal.Decimal(qx[index])
            tmp_lx = decimal.Decimal(lx[index])
        else:
            # 計算lx: 計算后，保證數(shù)據(jù)小數(shù)位為12位，與返回類型的設(shè)置小數(shù)位保持一致
            tmp_lx = (tmp_lx * (1 - tmp_qx)).quantize(decimal.Decimal('0.000000000000'))
            tmp_qx = decimal.Decimal(qx[index])
    return tmp_lx
if __name__ == '__main__':
    # 1) 創(chuàng)建 SparkSession 對象，此對象連接 hive
    spark = SparkSession.builder.master('local[*]') \
        .appName('insurance_main') \
        .config('spark.sql.shuffle.partitions', 4) \
        .config('spark.sql.warehouse.dir', 'hdfs://node1:8020/user/hive/warehouse') \
        .config('hive.metastore.uris', 'thrift://node1:9083') \
        .enableHiveSupport() \
        .getOrCreate()
    # 注冊UDAF 支持在SQL中使用
    spark.udf.register('udaf_lx', udaf_lx)
    # 2) 編寫SQL 執(zhí)行
    excuteSQLFile(spark, '_04_insurance_dw_prem_std.sql')

然后跑起來就報了以下錯誤：

Traceback (most recent call last):
  File "/root/anaconda3/lib/python3.8/site-packages/pyspark/sql/types.py", line 835, in _parse_datatype_string
    return from_ddl_datatype(s)
  File "/root/anaconda3/lib/python3.8/site-packages/pyspark/sql/types.py", line 827, in from_ddl_datatype
    sc._jvm.org.apache.spark.sql.api.python.PythonSQLUtils.parseDataType(type_str).json())
AttributeError: 'NoneType' object has no attribute '_jvm'
During handling of the above exception, another exception occurred:
Traceback (most recent call last):
  File "/root/anaconda3/lib/python3.8/site-packages/pyspark/sql/types.py", line 839, in _parse_datatype_string
    return from_ddl_datatype("struct<%s>" % s.strip())
  File "/root/anaconda3/lib/python3.8/site-packages/pyspark/sql/types.py", line 827, in from_ddl_datatype
    sc._jvm.org.apache.spark.sql.api.python.PythonSQLUtils.parseDataType(type_str).json())
AttributeError: 'NoneType' object has no attribute '_jvm'
During handling of the above exception, another exception occurred:
Traceback (most recent call last):
  File "/root/anaconda3/lib/python3.8/site-packages/pyspark/sql/types.py", line 841, in _parse_datatype_string
    raise e
  File "/root/anaconda3/lib/python3.8/site-packages/pyspark/sql/types.py", line 831, in _parse_datatype_string
    return from_ddl_schema(s)
  File "/root/anaconda3/lib/python3.8/site-packages/pyspark/sql/types.py", line 823, in from_ddl_schema
    sc._jvm.org.apache.spark.sql.types.StructType.fromDDL(type_str).json())
AttributeError: 'NoneType' object has no attribute '_jvm'

我左思右想，百思不得騎姐，嗐，跑去看 types.py里面的type類型，以為我的 udaf_lx 函數(shù)的裝飾器里面的 ‘decimal(17,12)’ 類型錯了，但是一看，好家伙，types.py 里面的774行

_FIXED_DECIMAL = re.compile(r"decimal\(\s*(\d+)\s*,\s*(-?\d+)\s*\)")

這是能匹配上的，沒道理??！

原因分析及解決方案：

然后再往回看報錯的信息的最后一行：

AttributeError: 'NoneType' object has no attribute '_jvm'

竟然是空對象沒有_jvm這個屬性！

一拍腦瓜子，得了，pyspark的SQL 在執(zhí)行的時候，需要用到 JVM ，而運行pyspark的時候，需要先要為spark提供環(huán)境，也就說，內(nèi)存中要有SparkSession對象，而python在執(zhí)行的時候，是從上往下，將方法加載到內(nèi)存中，在加載自定義的UDAF函數(shù)時，由于有裝飾器@F.pandas_udf的存在 , F 則是pyspark.sql.functions, 此時加載自定義的UDAF到內(nèi)存中，需要有SparkSession的環(huán)境提供JVM，而此時的內(nèi)存中尚未有SparkSession環(huán)境！因此，將自定義的UDAF 函數(shù)挪到 if __name__ == '__main__': 創(chuàng)建完SparkSession的后面，如下:

import decimal
import os
import pandas as pd
from pyspark.sql import SparkSession
from pyspark.sql import functions as F
os.environ['SPARK_HOME'] = '/export/server/spark'
os.environ["PYSPARK_PYTHON"] = "/root/anaconda3/bin/python"
os.environ["PYSPARK_DRIVER_PYTHON"] = "/root/anaconda3/bin/python"
if __name__ == '__main__':
    # 1) 創(chuàng)建 SparkSession 對象，此對象連接 hive
    spark = SparkSession.builder.master('local[*]') \
        .appName('insurance_main') \
        .config('spark.sql.shuffle.partitions', 4) \
        .config('spark.sql.warehouse.dir', 'hdfs://node1:8020/user/hive/warehouse') \
        .config('hive.metastore.uris', 'thrift://node1:9083') \
        .enableHiveSupport() \
        .getOrCreate()
    @F.pandas_udf('decimal(17,12)')
    def udaf_lx(qx: pd.Series, lx: pd.Series) -> decimal:
        # 初始值 也一定是decimal類型
        tmp_qx = decimal.Decimal(0)
        tmp_lx = decimal.Decimal(0)
        for index in range(0, qx.size):
            if index == 0:
                tmp_qx = decimal.Decimal(qx[index])
                tmp_lx = decimal.Decimal(lx[index])
            else:
                # 計算lx: 計算后，保證數(shù)據(jù)小數(shù)位為12位，與返回類型的設(shè)置小數(shù)位保持一致
                tmp_lx = (tmp_lx * (1 - tmp_qx)).quantize(decimal.Decimal('0.000000000000'))
                tmp_qx = decimal.Decimal(qx[index])
        return tmp_lx
    # 注冊UDAF 支持在SQL中使用
    spark.udf.register('udaf_lx', udaf_lx)
    # 2) 編寫SQL 執(zhí)行
    excuteSQLFile(spark, '_04_insurance_dw_prem_std.sql')

運行結(jié)果如圖：