一、UDF（User Defined Function）：spark SQL中使用者自定義函式，用法和spark SQL中的內建函式類似；是saprk SQL中內建函式無法滿足要求，使用者根據業務需求自定義的函式。

二、UDF使用分為兩步：

（1）自定義UDF類，根據業務需要，實現UDF1/2/3....22中的介面之一，其中UDF後跟的數字，
比如UDF1、UDF2；表示輸入引數的個數，1表示有一個入參，2表示有兩個入參，
最多可傳入22個輸入引數。如：
public class 自定義UDF類名 implements UDF3<String,String,String,String>{
    @override
    public String call(String p1,String p2,String p3)throw Exception{}
} 
其中UDF3中前三個引數表示輸入引數的資料型別，最後的引數用於指定call方法的返回值型別
（2）註冊UDF函式：
//建立配置
SparkConf conf = new SparkConf()
         .setAppName("openwindowFunctionDemo");
//建立提交spark應用的叢集入口類物件
JavaSparkContext sc = new JavaSparkContext(conf);
//建立SQLContext物件
SQLContext sqlContext = new SQLContext(sc.sc());
//註冊getJsonData自定義函式
/**
* register方法三個引數中，第一個引數指定自定義函式名，
* 第二個引數傳入一個自定義函式物件；第三個引數指定返回值型別
*
*/
sqlContext.udf().register("getJsonData",new GetJsonDataUDF(), DataTypes.StringType);
（3）通過sqlContext.sql(sql語句)使用上述註冊的UDF函式
 

三、具體案例：

 /**
         * 假如spark sql註冊過一張臨時表user，其資料格式如下
         * id       userInfo
         * 001      {'username':'tom','age':'12','sex':'男'}
         * 002      {'username':'henry','age':'18','sex':'男'}
         * 003      {'username':'jack','age':'32','sex':'男'}
         * 及user表中userInfo儲存的資料格式為json格式，如果需要單獨獲取userInfo裡的username；
         * spark sql內建的函式無法滿足我們的需要，這時就可以通過自定函式來實現上述需求
         */
（1）自定義UDF類：
public class GetJsonDataUDF implements UDF2<String,String,String>{
    @Override
    public String call(String json, String field) throws Exception {
        //構建json物件
        JSONObject jsonObject = JSONObject.parseObject(json);
        //根據欄位名獲取其對應的value
        return jsonObject.getString(field);
    }

}
（2）註冊UDF函式並使用
  
 

public class Test {
    public static void main(String[] args) {
        //建立配置
        SparkConf conf = new SparkConf()
                .setAppName("openwindowFunctionDemo");

        //建立提交spark應用的叢集入口類物件
        JavaSparkContext sc = new JavaSparkContext(conf);
        //建立SQLContext物件
        SQLContext sqlContext = new SQLContext(sc.sc());
        //註冊udf函式
        sqlContext.udf().register("getJsonData",new GetJsonDataUDF(), DataTypes.StringType);

        //構建臨時資料
        List<String> userList = new ArrayList<String>();
        userList.add("001 {'username':'tom','age':'12','sex':'男'}");
        userList.add("002 {'username':'henry','age':'18','sex':'男'}");
        userList.add("003 {'username':'jack','age':'32','sex':'男'}");

        JavaRDD<String> rdd = sc.parallelize(userList);
        //將原始資料轉化為格式：<id,userInfo>的形式
       JavaRDD<Row> userRDD = rdd.map(
               new Function<String, Row>() {
                   @Override
                   public Row call(String s) throws Exception {
                       //切分資料
                       String[] datas = s.split(" ");
                       String id = datas[0];
                       String userInfo = datas[1];
                       return RowFactory.create(id,userInfo);
                   }
               }
       );
       //註冊臨時表
       List<StructField> fields = new ArrayList<StructField>();
       fields.add(DataTypes.createStructField("id",DataTypes.StringType,true));
       fields.add(DataTypes.createStructField("userInfo",DataTypes.StringType,true));
       StructType schema = DataTypes.createStructType(fields);
       DataFrame dataFrame = sqlContext.createDataFrame(userRDD, schema);

       dataFrame.registerTempTable("tmp_user");

        //定義sql
        String sql = "SELECT id,"
                +"getJsonData(userInfo,'username')"
                +"FROM tmp_user";
        //開始查詢
        DataFrame res = sqlContext.sql(sql);
        res.show();
        sc.stop();
    }
}