案例说明flink的udf

public class HashCode extends ScalarFunction {
 private int factor = 12;
 public HashCode(int factor) {
     this.factor = factor;
 }
 public int eval(String s) {
   return s.hashCode() * factor;
 }
}
BatchTableEnvironment tableEnv = TableEnvironment.getTableEnvironment(env);
// register the function
tableEnv.registerFunction("hashCode", new HashCode(10));
// use the function in Java Table API
myTable.select("string, string.hashCode(), hashCode(string)");
// use the function in SQL API

tableEnv.sqlQuery("SELECT string, HASHCODE(string) FROM MyTable");

public static class TimestampModifier extends ScalarFunction {
 public long eval(long t) {
   return t % 1000;
 }
 public TypeInformation<?> getResultType(signature: Class<?>[]) {
   return Types.TIMESTAMP;
 }
}

// The generic type "Tuple2<String, Integer>" determines the schema of the returned table as (String, Integer).

public class Split extends TableFunction<Tuple2<String, Integer>> {

   private String separator = " ";
   public Split(String separator) {
       this.separator = separator;
   }
   public void eval(String str) {
       for (String s : str.split(separator)) {
           // use collect(...) to emit a row
           collect(new Tuple2<String, Integer>(s, s.length()));
       }
   }
}
BatchTableEnvironment tableEnv = TableEnvironment.getTableEnvironment(env);
Table myTable = ...         // table schema: [a: String]
// Register the function.
tableEnv.registerFunction("split", new Split("#"));
// Use the table function in the Java Table API. "as" specifies the field names of the table.
myTable.join("split(a) as (word, length)").select("a, word, length");

myTable.leftOuterJoin("split(a) as (word, length)").select("a, word, length");

// Use the table function in SQL with LATERAL and TABLE keywords.
// CROSS JOIN a table function (equivalent to "join" in Table API).
tableEnv.sqlQuery("SELECT a, word, length FROM MyTable, LATERAL TABLE(split(a)) as T(word, length)");
// LEFT JOIN a table function (equivalent to "leftOuterJoin" in Table API).
tableEnv.sqlQuery("SELECT a, word, length FROM MyTable LEFT JOIN LATERAL TABLE(split(a)) as T(word, length) ON TRUE");

public class CustomTypeSplit extends TableFunction<Row> {
   public void eval(String str) {
       for (String s : str.split(" ")) {
          Row row = new Row(2);
           row.setField(0, s);
           row.setField(1, s.length);
           collect(row);
       }
   }
   @Override
   public TypeInformation<Row> getResultType() {
       return Types.ROW(Types.STRING(), Types.INT());
   }
}

/**
* Accumulator for WeightedAvg.
*/
public static class WeightedAvgAccum {
   public long sum = 0;
   public int count = 0;
}
/**
* Weighted Average user-defined aggregate function.
*/
public static class WeightedAvg extends AggregateFunction<Long, WeightedAvgAccum> {
   @Override
   public WeightedAvgAccum createAccumulator() {
    return new WeightedAvgAccum();
   }
   @Override
   public Long getValue(WeightedAvgAccum acc) {
       if (acc.count == 0) {
           return null;
       } else {
           return acc.sum / acc.count;
       }
   }
   public void accumulate(WeightedAvgAccum acc, long iValue, int iWeight) {
       acc.sum += iValue * iWeight;
       acc.count += iWeight;
   }
   public void retract(WeightedAvgAccum acc, long iValue, int iWeight) {
       acc.sum -= iValue * iWeight;
       acc.count -= iWeight;
   }
   public void merge(WeightedAvgAccum acc, Iterable<WeightedAvgAccum> it) {
       Iterator<WeightedAvgAccum> iter = it.iterator();
       while (iter.hasNext()) {
           WeightedAvgAccum a = iter.next();
           acc.count += a.count;
           acc.sum += a.sum;
       }
   }
   public void resetAccumulator(WeightedAvgAccum acc) {
       acc.count = 0;
       acc.sum = 0L;
   }
}
// register function
StreamTableEnvironment tEnv = ...
tEnv.registerFunction("wAvg", new WeightedAvg());
// use function

tEnv.sqlQuery("SELECT user, wAvg(points, level) AS avgPoints FROM userScores GROUP BY user");

import org.apache.flink.table.functions.FunctionContext;
import org.apache.flink.table.functions.ScalarFunction;
import redis.clients.jedis.Jedis;
public class HashCode extends ScalarFunction {
   private int factor = 12;
   Jedis jedis = null;
   public HashCode() {
       super();
   }
   @Override
   public void open(FunctionContext context) throws Exception {
       super.open(context);
       String redisHost = context.getJobParameter("redis.host","localhost");
       int redisPort = Integer.valueOf(context.getJobParameter("redis.port","6379"));
       jedis = new Jedis(redisHost,redisPort);
   }

   @Override
   public void close() throws Exception {
       super.close();
       jedis.close();
   }

   public HashCode(int factor) {
       this.factor = factor;
   }

   public int eval(int s) {
       s = s % 3;
       if(s == 2)
           return Integer.valueOf(jedis.get(String.valueOf(s)));
       else
           return 0;
   }
}

ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
BatchTableEnvironment tableEnv = TableEnvironment.getTableEnvironment(env);
// set job parameter
Map<String,String> hashmap = new HashMap<>();
        hashmap.put("redis.host","localhost");
        hashmap.put("redis.port","6379");
        ParameterTool parameter = ParameterTool.fromMap(hashmap);
        exeEnv.getConfig().setGlobalJobParameters(parameter);
// register the function
tableEnv.registerFunction("hashCode", new HashCode());
// use the function in Java Table API
myTable.select("string, string.hashCode(), hashCode(string)");
// use the function in SQL
tableEnv.sqlQuery("SELECT string, HASHCODE(string) FROM MyTable");