前言
SpEL表达式与Flink fiter结合可以实现基于表达式的灵活动态过滤。有关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表达式，实现日期的比较过滤
String spel="compareDate(#row.getField(2), \"2016-10-24 00:48:36\")==0";
//实现对数字的过滤
// spel="#row.getField(3)>33";
SingleOutputStreamOperator<Row> filterStream = source.filter(new FilterSpelFunction(spel));
filterStream.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;
}
三、FilterFunction实现
import org.apache.flink.api.common.functions.RichFilterFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.types.Row;
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 spel.demo.util.SpelMethodUtil;
/**
* 基于spel 表达式的过滤
*/
public class FilterSpelFunction extends RichFilterFunction<Row> {
private static final Logger logger = LoggerFactory.getLogger(FilterSpelFunction.class);
private transient Expression exp;
private String filterExpr;
public FilterSpelFunction(String filterSpel) {
filterExpr=filterSpel;
logger.info("filterExpr:{}",filterExpr);
}
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
SpelExpressionParser parser = new SpelExpressionParser();
exp = parser.parseExpression(filterExpr);
}
@Override
public boolean filter(Row row) throws Exception {
try {
//注册自定义函数类
StandardEvaluationContext conetxt = new StandardEvaluationContext(new SpelMethodUtil());
//设置变量
conetxt.setVariable("row",row);
Boolean value = exp.getValue(conetxt, Boolean.class);
if (value == null) {
logger.error("表达式结果为null");
throw new Exception("表达式结果为null");
}
return value;
}catch (Exception e){
logger.error("filter 异常", e);
throw e;
}
}
}
自定义函数类文章来源：https://www.toymoban.com/news/detail-824401.html
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;
}
}
总结
以上只是简单的示例，在实际应用中可以将过滤表达式放到数据库，将过滤规则放入缓存定时刷新。大家可以根据实际需求进行扩展。文章来源地址https://www.toymoban.com/news/detail-824401.html
到了这里，关于Spring SpEL在Flink中的应用-与Filter结合实现数据动态分流的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！
