Spring SpEL在Flink中的应用-与Filter结合实现数据动态分流

2024-01-25 21:28

本文主要是介绍Spring SpEL在Flink中的应用-与Filter结合实现数据动态分流,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

文章目录

  • 前言
  • 一、POM依赖
  • 二、主函数代码示例
  • 三、FilterFunction实现
  • 总结


前言

SpEL表达式与Flink fiter结合可以实现基于表达式的灵活动态过滤。有关SpEL表达式的使用请参考Spring SpEL在Flink中的应用-SpEL详解
可以将过滤规则放入数据库,根据不同的数据设置不同的过滤表达式,从而实现只需修改过滤表达式不用修改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);}@Overridepublic void open(Configuration parameters) throws Exception {super.open(parameters);SpelExpressionParser parser = new SpelExpressionParser();exp = parser.parseExpression(filterExpr);}@Overridepublic 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;}}
}

自定义函数类


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;}
}

总结

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

这篇关于Spring SpEL在Flink中的应用-与Filter结合实现数据动态分流的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/644672

相关文章

C语言函数递归实际应用举例详解

《C语言函数递归实际应用举例详解》程序调用自身的编程技巧称为递归,递归做为一种算法在程序设计语言中广泛应用,:本文主要介绍C语言函数递归实际应用举例的相关资料,文中通过代码介绍的非常详细,需要的朋... 目录前言一、递归的概念与思想二、递归的限制条件 三、递归的实际应用举例(一)求 n 的阶乘(二)顺序打印

idea maven编译报错Java heap space的解决方法

《ideamaven编译报错Javaheapspace的解决方法》这篇文章主要为大家详细介绍了ideamaven编译报错Javaheapspace的相关解决方法,文中的示例代码讲解详细,感兴趣的... 目录1.增加 Maven 编译的堆内存2. 增加 IntelliJ IDEA 的堆内存3. 优化 Mave

Java String字符串的常用使用方法

《JavaString字符串的常用使用方法》String是JDK提供的一个类,是引用类型,并不是基本的数据类型,String用于字符串操作,在之前学习c语言的时候,对于一些字符串,会初始化字符数组表... 目录一、什么是String二、如何定义一个String1. 用双引号定义2. 通过构造函数定义三、St

springboot filter实现请求响应全链路拦截

《springbootfilter实现请求响应全链路拦截》这篇文章主要为大家详细介绍了SpringBoot如何结合Filter同时拦截请求和响应,从而实现​​日志采集自动化,感兴趣的小伙伴可以跟随小... 目录一、为什么你需要这个过滤器?​​​二、核心实现:一个Filter搞定双向数据流​​​​三、完整代码

SpringBoot利用@Validated注解优雅实现参数校验

《SpringBoot利用@Validated注解优雅实现参数校验》在开发Web应用时,用户输入的合法性校验是保障系统稳定性的基础,​SpringBoot的@Validated注解提供了一种更优雅的解... 目录​一、为什么需要参数校验二、Validated 的核心用法​1. 基础校验2. php分组校验3

Python实现AVIF图片与其他图片格式间的批量转换

《Python实现AVIF图片与其他图片格式间的批量转换》这篇文章主要为大家详细介绍了如何使用Pillow库实现AVIF与其他格式的相互转换,即将AVIF转换为常见的格式,比如JPG或PNG,需要的小... 目录环境配置1.将单个 AVIF 图片转换为 JPG 和 PNG2.批量转换目录下所有 AVIF 图

Java Predicate接口定义详解

《JavaPredicate接口定义详解》Predicate是Java中的一个函数式接口,它代表一个判断逻辑,接收一个输入参数,返回一个布尔值,:本文主要介绍JavaPredicate接口的定义... 目录Java Predicate接口Java lamda表达式 Predicate<T>、BiFuncti

Pydantic中model_validator的实现

《Pydantic中model_validator的实现》本文主要介绍了Pydantic中model_validator的实现,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价... 目录引言基础知识创建 Pydantic 模型使用 model_validator 装饰器高级用法mo

Spring Security基于数据库的ABAC属性权限模型实战开发教程

《SpringSecurity基于数据库的ABAC属性权限模型实战开发教程》:本文主要介绍SpringSecurity基于数据库的ABAC属性权限模型实战开发教程,本文给大家介绍的非常详细,对大... 目录1. 前言2. 权限决策依据RBACABAC综合对比3. 数据库表结构说明4. 实战开始5. MyBA

AJAX请求上传下载进度监控实现方式

《AJAX请求上传下载进度监控实现方式》在日常Web开发中,AJAX(AsynchronousJavaScriptandXML)被广泛用于异步请求数据,而无需刷新整个页面,:本文主要介绍AJAX请... 目录1. 前言2. 基于XMLHttpRequest的进度监控2.1 基础版文件上传监控2.2 增强版多