pyspark dataframe的自定义函数处理 udf方法及空值异常_pyspark 判断空值_头发又黑又密的博客

使用udf方法对dataframe的列使用自定义函数

from pyspark.sql.types import StringType
from pyspark.sql.types import DoubleType
from pyspark.sql.functions import udf
def avg_score(score, people):
	try:
		if int(people) != 0:
			return int(score) / int(people)
		else:
			return 0
	except:
		return 0
func = udf(avg_score, Doubletype())
#  此处省略df的获取步骤
df = df.select("id", "total_score", "total_people")
df = df.withColumn("avg_score",func(df.total_score, df.total_people))
此时，我们通过自定义函数avg_score就可以根据两个表项得到我们需要的数据，再通过df.drop(“total_score”).drop(“total_people”) 丢弃不需要再使用的两项，或者通过df.select(“id”, “avg_score”)选取我们需要的两项去进行存表操作等。 
空值异常情况，表中null，none等数据未进入自定义函数处理。
 
由于 udf返回类型是DoubleType类型时，为null和none的数据行不会被传入自定义函数执行，所以看起来出现了错误。 
解决办法：可先使用返回类型为StringType的自定义函数，在函数中加入对null的处理（返回0），再使用DoubleType类型的自定义函数将值转为DoubleType
                    使用udf方法对dataframe的列使用自定义函数from pyspark.sql.types import StringTypefrom pyspark.sql.types import DoubleTypefrom pyspark.sql.functions import udfdef avg_score(score, people):	try:		if int(people)...
UDF(User-Defined-Function)，用户自定义函数对数据进行处理。
STEP1：创建UDF函数
        ①自定义UDF需要继承org.apache.hadoop.hive.ql.UDF
        ②需要实现evaluate函数
STEP2：将udf打jar包，并上传到hive server上
STEP3：进入hi...
%pyspark
data = [(1,"火男",6000),(1,"流浪法师",7000),(2,"盖伦",10000),(2,"皇子",8000)]
df = spark.createDataFrame(data,["属性","角色名称","战斗力"])
df.show()
二、pyspark注册udf
%pyspark
#注册udf
import pandas as pd
from pyspark.shell import sqlContext
from pyspark.sql import SparkSession
from pyspark.sql import functions as func
from pyspark.sql.types import StringType, ArrayType, StructType, StructField
创建spark
spark = SparkSession.builder.enableHiveSupport().
				在使用PySpark的DataFrame处理数据时，会遇到需要用自定义函数（user defined function, udf）对某一列或几列进行运算，生成新列的情况。PySpark的udf使用方法如下。
先初始化SparkContext和SparkSession
from pyspark import SparkConf
from pyspark import SparkContext
fro...
白色提示：
jarstore/testudf.jar does not exist
Query returned non-zero code: 1, cause: jarstore/testudf.jar does not exist
add jar hdfs://hadoop100:9000/jarstore/testudf.jar;
create temporary function jar_test as "cn.k
				PySpark UDF概念引出
在pandas中自定义函数，通过遍历行的方式，便捷实现工程师的需求。但是对于数据量较大的数据处理，会出现速度过慢甚至超内存的问题。Spark作为替代pandas处理海量数据的工具，参照 pandas udf 定义了名为PandasUDFType的类，通过自定义函数的方式spark处理数据的灵活度和高效率有很大亮点。
从spark 1.3到2.3udf函数有row-a...