pyspark自定義UDAF函數調用報錯如何解決

這篇文章主要講解了“pyspark自定義UDAF函數調用報錯如何解決”，文中的講解內容簡單清晰，易于學習與理解，下面請大家跟著小編的思路慢慢深入，一起來研究和學習“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函數調用報錯如何解決”的內容了，經過本文的學習后，相信大家對pyspark自定義UDAF函數調用報錯如何解決這一問題有了更深刻的體會，具體使用情況還需要大家實踐驗證。這里是億速云，小編將為大家推送更多相關知識點的文章，歡迎關注！