pyspark自定義UDAF函數呼叫報錯問題解決

2022-06-08 22:01:16

問題場景：

在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函數報錯的資料請關注it145.com其它相關文章！

pyspark自定義UDAF函數呼叫報錯問題解決

目錄

問題場景：

問題描述

原因分析及解決方案：

熱門文章