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其它相關文章!
推薦閱讀:
- pycharm利用pyspark遠程連接spark集群的實現
- Python安裝spark的詳細過程
- pyspark對Mysql數據庫進行讀寫的實現
- pyspark創建DataFrame的幾種方法
- 運行獨立 pyspark 時出現 Windows 錯誤解決辦法