---

前言

SpEL表达式与Flink FlatMapFunction或MapFunction结合可以实现基于表达式的简单动态计算。有关SpEL表达式的使用请参考Spring SpEL在Flink中的应用-SpEL详解。
可以将计算表达式放入数据库，对数据进行计算处理，从而实现只需修改表达式不用修改Flink代码就能实现数据计算。对于基于Flink进行数据计算平台建设会起到事半功倍的效果。

---

一、POM依赖

首先在 pom.xml 中加入依赖：

<dependency><groupId>org.springframework</groupId><artifactId>spring-expression</artifactId><version>5.2.0.RELEASE</version>
</dependency>​

二、主函数代码示例

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.types.Row;import java.text.SimpleDateFormat;public class FlinkSpelFilterDemo {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();Row row=Row.of("张三","001",getTimestamp("2016-10-24 21:59:06"),23);Row row2=Row.of("张三","002",getTimestamp("2016-10-24 21:50:06"),33);Row row3=Row.of("张三","003",getTimestamp("2016-10-24 21:51:06"),43);Row row4=Row.of("李四","004",getTimestamp("2016-10-24 21:50:56"),13);Row row5=Row.of("李四","005",getTimestamp("2016-10-24 00:48:36"),53);Row row6=Row.of("李四","006",getTimestamp("2016-10-24 00:48:36"),34);Row row7=Row.of("李四","007",getTimestamp("2016-10-24 00:48:36"),23);Row row8=Row.of("李四","008",getTimestamp("2016-10-24 00:48:36"),26);Row row9=Row.of("李四","009",getTimestamp("2016-10-24 00:48:36"),63);DataStreamSource<Row> source =env.fromElements(row,row2,row3,row4,row5,row6,row7,row8,row9);//spel表达式 //json串数据略...................JSONObject spelConfig=".................................";SingleOutputStreamOperator<Row> stream = source.flatmap(new FlatMapExprFunction (spelConfig));stream .print();env.execute();}private static java.sql.Timestamp getTimestamp(String str) throws Exception {
//		String string = "2016-10-24 21:59:06";SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");java.util.Date date=sdf.parse(str);java.sql.Timestamp s = new java.sql.Timestamp(date.getTime());return s;}

三、RichFlatMapFunction实现

import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.types.Row;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.expression.Expression;
import org.springframework.expression.spel.standard.SpelExpressionParser;
import org.springframework.expression.spel.support.StandardEvaluationContext;import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;/*** @author gaowc* 基于表达式计算的flatmap*/
public class FlatMapExprFunction extends RichFlatMapFunction<Row, Row> {private static final Logger logger = LoggerFactory.getLogger(FlatMapExprFunction.class);/*** 列名和列的索引map key:列名 value:列索引*/private Map<String, Integer> columnIndexMap;/*** key:输出字段名 value：表达式对象*/private transient Map<String,Expression> expMap;private Integer size;private List<JSONObject> outputColumnList;public FlatMapExprFunction(JSONObject conf){List<JSONObject > columnList = conf.getList(Constants.COLUMN);columnIndexMap = TransformUtil.getColumnIndexMap(columnList);List<JSONObject > outputColumns = conf.getList(Constants.OUTPUT_COLUMN);//将表达式中的占位符替换为row.getField(x)size = outputColumns.size();outputColumnList = new ArrayList<>();for (JSONObject col:outputColumns) {String expr = col.getString("expr");if(StringUtils.isNotBlank(expr)){ExprTokenParser tokenParser = new ExprTokenParser("#{","}",new ColumnTokenHandler(columnIndexMap));String newExpr=tokenParser.parse(expr);logger.info("expr: {} newExpr {}",expr,newExpr);col.set("expr",newExpr);}outputColumnList.add(col);}}@Overridepublic void open(Configuration parameters) throws Exception {super.open(parameters);expMap = new HashMap<>();//初始化expMapfor (JSONObject col:outputColumnList) {logger.info("open col:{}",col);String expr = col.getString("expr");if(StringUtils.isNotBlank(expr)){SpelExpressionParser parser = new SpelExpressionParser();Expression expression = parser.parseExpression(expr);expMap.put(col.getString(Constants.COLUMN_NAME),expression);}}}@Overridepublic void flatMap(Row row, Collector<Row> collector) throws Exception {Row outputRow=new Row(size);//注册自定义函数StandardEvaluationContext conetxt = new StandardEvaluationContext(new SpelMethodUtil());conetxt.setVariable("row",row);for (int i = 0; i < size; i++) {JSONObject col = outputColumnList.get(i);String colName = col.getString(Constants.COLUMN_NAME);Expression expression = expMap.get(colName);Object value = null;if(expression!=null){value = expression.getValue(conetxt);if(value!=null){logger.info("expression.getValue :{}  class {}",value,value.getClass() );}}else{value=row.getField(columnIndexMap.get(colName));}outputRow.setField(i,value);}collector.collect(outputRow);}}

自定义函数类

import org.apache.commons.lang3.StringUtils;import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;public class SpelMethodUtil {public static final String TIMESTAMP_FORMAT = "yyyy-MM-dd HH:mm:ss";public static final String DATE_FORMAT = "yyyy-MM-dd";public static final String TIME_FORMAT = "HH:mm:ss";public static Integer compareDate(Date date, String strDate){Integer result;if(date==null&& StringUtils.isBlank(strDate)){return 0;}else{if(date==null || StringUtils.isBlank(strDate)){return -2;}}String trimDate=strDate.trim();String format = findFormat(trimDate);Date date2 = stringToDate(trimDate, format);result=date.compareTo(date2);return result;}public static Integer compareDate(Date first, Date second){if(first==null&& second==null){return 0;}else{if(first==null || second==null){return -2;}}return first.compareTo(second);}public static Date stringToDate(String dateStr,String format){SimpleDateFormat sdf = new SimpleDateFormat(format);Date date=null;try {date= sdf.parse(dateStr);} catch (ParseException e) {e.printStackTrace();}return date;}/*** 查找与输入的字符型日期相匹配的format* @param strDate* @return*/public static String findFormat(String strDate){String result=null;String trimDate=strDate.trim();int len=trimDate.length();String dateRegex = "";if(len==TIMESTAMP_FORMAT.length()){dateRegex = "^\\d{4}-\\d{2}-\\d{2} \\d{2}:\\d{2}:\\d{2}$";if(trimDate.matches(dateRegex)){result=TIMESTAMP_FORMAT;}}else if(len==DATE_FORMAT.length()){dateRegex = "^\\d{4}-\\d{2}-\\d{2}$";if(trimDate.matches(dateRegex)){result=DATE_FORMAT;}}else if(len==TIME_FORMAT.length()){dateRegex = "^\\d{2}:\\d{2}:\\d{2}$";if(trimDate.matches(dateRegex)){result=TIME_FORMAT;}}else{throw  new RuntimeException("不可识别的日期格式！"+strDate);}return result;}public static Integer addAge(Integer age){return age+4;}
}

---

总结

以上只是简单的示例，在实际应用中可以将运算表达式放到数据库，将计算规则放入缓存定时刷新。大家可以根据实际需求进行扩展。