pyspark自定義UDAF函數調用報錯問題解決

問題場景:

在SparkSQL中,因為需要用到自定義的UDAF函數,所以用pyspark自定義瞭一個,但是遇到瞭一個問題,就是自定義的UDAF函數一直報

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

在此將解決過程記錄下來

問題描述

在新建的py文件中,先自定義瞭一個UDAF函數,然後在 if __name__ == '__main__': 中調用,死活跑不起來,一遍又一遍的對源碼,看起來自定義的函數也沒錯:過程如下:

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: 計算後,保證數據小數位為12位,與返回類型的設置小數位保持一致
            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) 創建 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 執行
    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 函數的裝飾器裡面的 ‘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 在執行的時候,需要用到 JVM ,而運行pyspark的時候,需要先要為spark提供環境,也就說,內存中要有SparkSession對象,而python在執行的時候,是從上往下,將方法加載到內存中,在加載自定義的UDAF函數時,由於有裝飾器@F.pandas_udf的存在 , F 則是pyspark.sql.functions, 此時加載自定義的UDAF到內存中,需要有SparkSession的環境提供JVM,而此時的內存中尚未有SparkSession環境!因此,將自定義的UDAF 函數挪到 if __name__ == '__main__': 創建完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) 創建 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: 計算後,保證數據小數位為12位,與返回類型的設置小數位保持一致
                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 執行
    excuteSQLFile(spark, '_04_insurance_dw_prem_std.sql')

運行結果如圖:

至此,完美解決!更多關於pyspark自定義UDAF函數報錯的資料請關註WalkonNet其它相關文章!

推薦閱讀: