public class UDF {
public static void main(String[] args) {
SparkConf conf = new SparkConf().setMaster(“local”).setAppName(“udf”);
JavaSparkContext sc = new JavaSparkContext(conf);
SQLContext sqlContext = new SQLContext(sc);
JavaRDD parallelize = sc.parallelize(Arrays.asList(“zhangsan”,“lisi”,“wangwu”));
//将rdd转为rowrdd
JavaRDD rowRDD = parallelize.map(new Function() {
/**
*
*/
private static final long serialVersionUID = 1L;
@Override
public Row call(String s) throws Exception {
return RowFactory.create(s);
}
});
/**
* 动态创建Schema方式加载DF
/
List fields = new ArrayList();
fields.add(DataTypes.createStructField(“name”, DataTypes.StringType,true));
StructType schema = DataTypes.createStructType(fields);
DataFrame df = sqlContext.createDataFrame(rowRDD,schema);
df.registerTempTable(“user”);
/*
* 根据UDF函数参数的个数来决定是实现哪一个UDF UDF1，UDF2。。。。UDF1xxx
*/
sqlContext.udf().register(“StrLen”, new UDF1() {
private static final long serialVersionUID = 1L;
@Override
public Integer call(String t1) throws Exception {
return t1.length();
}
}, DataTypes.IntegerType);
sqlContext.sql(“select name ,StrLen(name) as length from user”).show();
// sqlContext.udf().register(“StrLen”,new UDF2() {
// private static final long serialVersionUID = 1L;
//
// @Override
// public Integer call(String t1, Integer t2) throws Exception {
// return t1.length()+t2;
// }
// } ,DataTypes.IntegerType );
// sqlContext.sql(“select name ,StrLen(name,10) as length from user”).show();
sc.stop();
}
}
