flink-connector-redis支持select查询

EN

1 项目介绍

基于bahir-flink二次开发,相对bahir调整的内容有:

1.使用Lettuce替换Jedis,同步读写改为异步读写,大幅度提升了性能 
2.增加了Table/SQL API,增加select/维表join查询支持
3.增加关联查询缓存(支持增量与全量)
4.增加支持整行保存功能,用于多字段的维表关联查询
5.增加限流功能,用于Flink SQL在线调试功能
6.增加支持Flink高版本(包括1.12,1.13,1.14+)
7.统一过期策略等
8.支持flink cdc删除及其它RowKind.DELETE
9.支持select查询

因bahir使用的flink接口版本较老,所以改动较大,开发过程中参考了腾讯云与阿里云两家产商的流计算产品,取两家之长,并增加了更丰富的功能。

注:redis不支持两段提交无法实现刚好一次语义。

2 使用方法:

2.1 工程直接引用

项目依赖Lettuce(6.2.1)及netty-transport-native-epoll(4.1.82.Final),如flink环境有这两个包,则使用flink-connector-redis-1.3.2.jar,
否则使用flink-connector-redis-1.4.1-jar-with-dependencies.jar。

<dependency><groupId>io.github.jeff-zou</groupId><artifactId>flink-connector-redis</artifactId><!-- 没有单独引入项目依赖Lettuce netty-transport-native-epoll依赖时 --><!--            <classifier>jar-with-dependencies</classifier>--><version>1.4.1</version>
</dependency>

2.2 自行打包

打包命令: mvn package -DskipTests,将生成的包放入flink lib中即可,无需其它设置。

2.3 使用示例

-- 创建redis表示例
create table redis_table (name varchar, age int) with ('connector'='redis', 'host'='10.11.69.176', 'port'='6379','password'='test123', 'redis-mode'='single','command'='set');
-- 写入  insert into redis_table select * from (values('test', 1));-- 查询  insert into redis_table select name,age + 1 from redis_table /*+ options('scan.key'='test') */create table gen_table (age int , level int, proctime as procTime()) with ('connector'='datagen','fields.age.kind' = 'sequence','fields.age.start' = '2','fields.age.end' = '2','fields.level.kind' = 'sequence','fields.level.start' = '10','fields.level.end' = '10'); -- 关联查询 
insert into redis_table select 'test', j.age + 10 from gen_table s left join redis_table  for system_time as of proctime as j
on j.name = 'test'

3 参数说明:

3.1 主要参数:

字段默认值类型说明
connector(none)Stringredis
host(none)StringRedis IP
port6379IntegerRedis 端口
passwordnullString如果没有设置,则为 null
database0Integer默认使用 db0
timeout2000Integer连接超时时间,单位 ms,默认 1s
cluster-nodes(none)String集群ip与端口,当redis-mode为cluster时不为空,如:10.11.80.147:7000,10.11.80.147:7001,10.11.80.147:8000
command(none)String对应上文中的redis命令
redis-mode(none)Integermode类型: single cluster sentinel
lookup.cache.max-rows-1Integer查询缓存大小,减少对redis重复key的查询
lookup.cache.ttl-1Integer查询缓存过期时间,单位为秒, 开启查询缓存条件是max-rows与ttl都不能为-1
lookup.cache.load-allfalseBoolean开启全量缓存,当命令为hget时,将从redis map查询出所有元素并保存到cache中,用于解决缓存穿透问题
max.retries1Integer写入失败重试次数
value.data.structurecolumnStringcolumn: value值来自某一字段 (如, set: key值取自DDL定义的第一个字段, value值取自第二个字段)
row: 将整行内容保存至value并以’\01’分割
set.if.absentfalseBoolean在key不存在时才写入,只对set hset有效
io.pool.size(none)IntegerLettuce内netty的io线程池大小,默认情况下该值为当前JVM可用线程数,并且大于2
event.pool.size(none)IntegerLettuce内netty的event线程池大小 ,默认情况下该值为当前JVM可用线程数,并且大于2
scan.key(none)String查询时redis key
scan.addition.key(none)String查询时限定redis key,如map结构时的hashfield
scan.range.start(none)Integer查询list结构时指定lrange start
scan.range.stop(none)Integer查询list结构时指定lrange start
scan.count(none)Integer查询set结构时指定srandmember count

3.1.1 command值与redis命令对应关系:

command值写入查询维表关联删除(Flink CDC等产生的RowKind.delete)
setsetgetgetdel
hsethsethgethgethdel
getsetgetgetdel
hsethsethgethgethdel
rpushrpushlrange
lpushlpushlrange
incrBy incrByFloatincrBy incrByFloatgetget写入相对值,如:incrby 2 -> incryby -2
hincrBy hincryByFloathincrBy hincryByFloathgethget写入相对值,如:hincrby 2 -> hincryby -2
zincrbyzincrbyzscorezscore写入相对值,如:zincrby 2 -> zincryby -2
saddsaddsrandmember 10srem
zaddzaddzscorezscorezrem
pfadd(hyperloglog)pfadd(hyperloglog)
publishpublish
zremzremzscorezscore
sremsremsrandmember 10
deldelgetget
hdelhdelhgethget
decrBydecrBygetget

注:为空表示不支持

3.1.2 value.data.structure = column(默认)

无需通过primary key来映射redis中的Key,直接由ddl中的字段顺序来决定Key,如:

create table sink_redis(username VARCHAR, passport VARCHAR)  with ('command'='set') 
其中username为key, passport为value.create table sink_redis(name VARCHAR, subject VARCHAR, score VARCHAR)  with ('command'='hset') 
其中name为map结构的key, subject为field, score为value.

3.1.3 value.data.structure = row

整行内容保存至value并以’\01’分割

create table sink_redis(username VARCHAR, passport VARCHAR)  with ('command'='set') 
其中username为key, username\01passport为value.create table sink_redis(name VARCHAR, subject VARCHAR, score VARCHAR)  with ('command'='hset') 
其中name为map结构的key, subject为field, name\01subject\01score为value.

3.2 sink时ttl相关参数

FieldDefaultTypeDescription
ttl(none)Integerkey过期时间(秒),每次sink时会设置ttl
ttl.on.time(none)Stringkey的过期时间点,格式为LocalTime.toString(), eg: 10:00 12:12:01,当ttl未配置时才生效
ttl.key.not.absentfalseboolean与ttl一起使用,当key不存在时才设置ttl

3.3 在线调试SQL时,用于限制sink资源使用的参数:

FieldDefaultTypeDescription
sink.limitfalseBoolean是否打开限制
sink.limit.max-num10000Integertaskmanager内每个slot可以写的最大数据量
sink.limit.interval100Stringtaskmanager内每个slot写入数据间隔 milliseconds
sink.limit.max-online30 * 60 * 1000LLongtaskmanager内每个slot最大在线时间, milliseconds

3.4 集群类型为sentinel时额外连接参数:

字段默认值类型说明
master.name(none)String主名
sentinels.info(none)String如:10.11.80.147:7000,10.11.80.147:7001,10.11.80.147:8000
sentinels.password(none)Stringsentinel进程密码

4 数据类型转换

flink typeredis row converter
CHARString
VARCHARString
StringString
BOOLEANString String.valueOf(boolean val)
boolean Boolean.valueOf(String str)
BINARYString Base64.getEncoder().encodeToString
byte[] Base64.getDecoder().decode(String str)
VARBINARYString Base64.getEncoder().encodeToString
byte[] Base64.getDecoder().decode(String str)
DECIMALString BigDecimal.toString
DecimalData DecimalData.fromBigDecimal(new BigDecimal(String str),int precision, int scale)
TINYINTString String.valueOf(byte val)
byte Byte.valueOf(String str)
SMALLINTString String.valueOf(short val)
short Short.valueOf(String str)
INTEGERString String.valueOf(int val)
int Integer.valueOf(String str)
DATEString the day from epoch as int
date show as 2022-01-01
TIMEString the millisecond from 0’clock as int
time show as 04:04:01.023
BIGINTString String.valueOf(long val)
long Long.valueOf(String str)
FLOATString String.valueOf(float val)
float Float.valueOf(String str)
DOUBLEString String.valueOf(double val)
double Double.valueOf(String str)
TIMESTAMPString the millisecond from epoch as long
timestamp TimeStampData.fromEpochMillis(Long.valueOf(String str))

5 使用示例:

  • 5.1 维表查询:

create table sink_redis(name varchar, level varchar, age varchar) with ( 'connector'='redis', 'host'='10.11.80.147','port'='7001', 'redis-mode'='single','password'='******','command'='hset');-- 先在redis中插入数据,相当于redis命令: hset 3 3 100 --
insert into sink_redis select * from (values ('3', '3', '100'));create table dim_table (name varchar, level varchar, age varchar) with ('connector'='redis', 'host'='10.11.80.147','port'='7001', 'redis-mode'='single', 'password'='*****','command'='hget', 'maxIdle'='2', 'minIdle'='1', 'lookup.cache.max-rows'='10', 'lookup.cache.ttl'='10', 'lookup.max-retries'='3');-- 随机生成10以内的数据作为数据源 --
-- 其中有一条数据会是: username = 3  level = 3, 会跟上面插入的数据关联 -- 
create table source_table (username varchar, level varchar, proctime as procTime()) with ('connector'='datagen',  'rows-per-second'='1',  'fields.username.kind'='sequence',  'fields.username.start'='1',  'fields.username.end'='10', 'fields.level.kind'='sequence',  'fields.level.start'='1',  'fields.level.end'='10');create table sink_table(username varchar, level varchar,age varchar) with ('connector'='print');insert intosink_table
selects.username,s.level,d.age
fromsource_table s
left join dim_table for system_time as of s.proctime as d ond.name = s.usernameand d.level = s.level;
-- username为3那一行会关联到redis内的值,输出为: 3,3,100	
  • 5.2 多字段的维表关联查询

很多情况维表有多个字段,本实例展示如何利用’value.data.structure’='row’写多字段并关联查询。

-- 创建表
create table sink_redis(uid VARCHAR,score double,score2 double )
with ( 'connector' = 'redis','host' = '10.11.69.176','port' = '6379','redis-mode' = 'single','password' = '****','command' = 'SET','value.data.structure' = 'row');  -- 'value.data.structure'='row':整行内容保存至value并以'\01'分割
-- 写入测试数据,score、score2为需要被关联查询出的两个维度
insert into sink_redis select * from (values ('1', 10.3, 10.1));-- 在redis中,value的值为: "1\x0110.3\x0110.1" --
-- 写入结束 ---- create join table --
create table join_table with ('command'='get', 'value.data.structure'='row') like sink_redis-- create result table --
create table result_table(uid VARCHAR, username VARCHAR, score double, score2 double) with ('connector'='print')-- create source table --
create table source_table(uid VARCHAR, username VARCHAR, proc_time as procTime()) with ('connector'='datagen', 'fields.uid.kind'='sequence', 'fields.uid.start'='1', 'fields.uid.end'='2')-- 关联查询维表,获得维表的多个字段值 --
insertintoresult_table
selects.uid,s.username,j.score, -- 来自维表j.score2 -- 来自维表
fromsource_table as s
join join_table for system_time as of s.proc_time as j onj.uid = s.uidresult:
2> +I[2, 1e0fe885a2990edd7f13dd0b81f923713182d5c559b21eff6bda3960cba8df27c69a3c0f26466efaface8976a2e16d9f68b3, null, null]
1> +I[1, 30182e00eca2bff6e00a2d5331e8857a087792918c4379155b635a3cf42a53a1b8f3be7feb00b0c63c556641423be5537476, 10.3, 10.1]
  • 5.3 DataStream查询方式

    示例代码路径: src/test/java/org.apache.flink.streaming.connectors.redis.datastream.DataStreamTest.java

    hset示例,相当于redis命令:hset tom math 150

      Configuration configuration = new Configuration();configuration.setString(REDIS_MODE, REDIS_CLUSTER);configuration.setString(REDIS_COMMAND, RedisCommand.HSET.name());RedisSinkMapper redisMapper = (RedisSinkMapper)RedisHandlerServices.findRedisHandler(RedisMapperHandler.class, configuration.toMap()).createRedisMapper(configuration);StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();GenericRowData genericRowData = new GenericRowData(3);genericRowData.setField(0, "tom");genericRowData.setField(1, "math");genericRowData.setField(2, "152");DataStream<GenericRowData> dataStream = env.fromElements(genericRowData, genericRowData);RedisSinkOptions redisSinkOptions =new RedisSinkOptions.Builder().setMaxRetryTimes(3).build();FlinkConfigBase conf =new FlinkSingleConfig.Builder().setHost(REDIS_HOST).setPort(REDIS_PORT).setPassword(REDIS_PASSWORD).build();RedisSinkFunction redisSinkFunction =new RedisSinkFunction<>(conf, redisMapper, redisSinkOptions, resolvedSchema);dataStream.addSink(redisSinkFunction).setParallelism(1);env.execute("RedisSinkTest");
  • 5.4 redis-cluster写入示例

    示例代码路径: src/test/java/org.apache.flink.streaming.connectors.redis.table.SQLInsertTest.java

    set示例,相当于redis命令: set test test11

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
EnvironmentSettings environmentSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
StreamTableEnvironment tEnv = StreamTableEnvironment.create(env, environmentSettings);String ddl = "create table sink_redis(username VARCHAR, passport VARCHAR) with ( 'connector'='redis', " +"'cluster-nodes'='10.11.80.147:7000,10.11.80.147:7001','redis- mode'='cluster','password'='******','command'='set')" ;tEnv.executeSql(ddl);
String sql = " insert into sink_redis select * from (values ('test', 'test11'))";
TableResult tableResult = tEnv.executeSql(sql);
tableResult.getJobClient().get()
.getJobExecutionResult()
.get();

6 解决问题联系我

https://github.com/jeff-zou/flink-connector-redis.git

7 开发与测试环境

ide: IntelliJ IDEA

code format: google-java-format + Save Actions

code check: CheckStyle

flink 1.12/1.13/1.14+

jdk1.8 Lettuce 6.2.1

8 如果需要flink 1.12版本支持,请切换到分支flink-1.12(注:1.12使用jedis)

<dependency><groupId>io.github.jeff-zou</groupId><artifactId>flink-connector-redis</artifactId><version>1.1.1-1.12</version>
</dependency>

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.mzph.cn/news/770801.shtml

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

flask_restful数据解析

参数验证也叫参数解析 Flask-Restful 插件提供了类似 WTForms 来验证提交的数据是否合法 的包&#xff0c;叫做 reqparse 。 # Flask_RESTFUl数据解析 from flask import Flask,render_template from flask_restful import Api,Resource from flask_restful.reqparse import …

项目3-留言板

1.创建项目 记得将project type改为maven 将需要的包引入其中 更改版本号 引入MYSQL相关包记得进行配置&#xff01;&#xff01;&#xff01; spring:datasource:url: jdbc:mysql://127.0.0.1:3306/mycnblog?characterEncodingutf8&useSSLfalseusername: rootpassword:…

深入解析Oracle数据库中的外连接 (OUTER JOIN)

在Oracle数据库中&#xff0c;外连接用于从两个或更多表的连接操作中检索那些在至少一个表中有匹配记录的所有记录。外连接可以确保即使在另一个表中找不到匹配项&#xff0c;指定表中的记录也会出现在结果集中。外连接主要有三种类型&#xff1a; 左外连接 (LEFT OUTER JOIN)&…

sql server用nest typeorm实现索引的方式

针对您提到的索引类型&#xff0c;下面是使用TypeORM库在SQL Server中实现不同类型的索引的代码示例&#xff1a; 普通索引&#xff1a; import { Entity, Column, Index } from typeorm;Entity() Index(idx_name, [name]) export class User {Column()name: string;Column()…

JavaScript快速入门笔记之一(基本概念)

JavaScript快速入门笔记之一&#xff08;基本概念&#xff09; 前端三大语言&#xff1a; HTML&#xff1a;专门编写网页内容的语言CSS&#xff1a;专门美化网页样式的语言JavaScript&#xff1a;专门编写网页交互的语言 名词解释&#xff1a; 交互&#xff1a;输入数据&#…

C++ MFC 只启动一个程序实例 唤醒之前的实例(完整源码)

初级代码游戏的专栏介绍与文章目录-CSDN博客 很多时候我们希望只允许启动一个程序实例&#xff0c;如果再次运行&#xff0c;就唤醒之前的实例。 目录 1 概述 2 相关技术介绍 2.1 互斥对象 2.2 查找窗口 2.3 唤醒窗口 1 概述 技术上并不难&#xff0c;涉及到以下几个技术…

用redis lua脚本实现时间窗分布式限流

需求背景&#xff1a; 限制某sql在30秒内最多只能执行3次 需求分析 微服务分布式部署&#xff0c;既然是分布式限流&#xff0c;首先自然就想到了结合redis的zset数据结构来实现。 分析对zset的操作&#xff0c;有几个步骤&#xff0c;首先&#xff0c;判断zset中符合rangeS…

Java服务器开发的日志:日志的作用和使用方法;日志文件的定期归档、清理和滚动策略;ELK(又称Elastic Stack)

Java服务器开发的日志 Java服务器开发&#xff0c;为什么要使用日志&#xff1f; 应该怎样使用日志&#xff1f; 在Java服务器开发中使用日志是非常关键的一个实践&#xff0c;原因如下&#xff1a; 本文所说的“日志”&#xff0c;主要是指程序运行时生成的技术日志&#xff…

#AngularJS#$sce.trustAsResourceUrl

$sce.trustAsResourceUrl 是 AngularJS&#xff08;一个旧版本的 Angular&#xff09;中的安全上下文&#xff08;Security Context&#xff09;方法&#xff0c;用于信任一个 URL&#xff0c;使其可以作为资源 URL 使用&#xff0c;而不会触发 AngularJS 的安全警告或阻止。 …

使用JMeter进行梯度压测

使用JMeter进行梯度压测 梯度压测配置如下&#xff1a; 使用线程:5&#xff0c;然后循环5000次&#xff0c;共2.5万个样本使用线程:10&#xff0c;然后循环5000次&#xff0c;共5万个样本使用线程:15&#xff0c;然后循环5000次&#xff0c;共7.5万个样本使用线程:20&#xff…

Redis中的事件

事件 概述 Redis服务器是一个事件驱动程序:服务器需要处理以下两类事件: 1.文件事件(file event):Redis服务器通过套接字与客户端(或者其他Redis服务器)进行连接&#xff0c;而文件事件就是服务器对套接字操作的抽象。服务器与客户端(或者其他服务器)的通信会产生相应的文件…

上位机图像处理和嵌入式模块部署(qmacvisual自定义插件代码分析)

【 声明&#xff1a;版权所有&#xff0c;欢迎转载&#xff0c;请勿用于商业用途。 联系信箱&#xff1a;feixiaoxing 163.com】 qmacvisual里面的第三方插件主要由两部分组成&#xff0c;一部分是ExtensionLibrary&#xff0c;也就是插件的容器&#xff0c;这个是官方提供的&a…

【机器学习】贝叶斯上篇(详解)

深入理解贝叶斯学习&#xff1a;核心原理及应用全解析 在机器学习的领域内&#xff0c;贝叶斯学习作为一种强大的框架&#xff0c;使我们能够在不确定性条件下进行预测和决策。贝叶斯学习源于托马斯贝叶斯的工作&#xff0c;提供了一种概率论的学习方法&#xff0c;与传统的频…

01 React新建开发环境

https://create-react-app.dev/docs/getting-started npx create-react-app my-appJSX使用表达式嵌入 function App() {const count 100;function getSelfName() {return "SelfName"}return (<div>Hello World!<div>{This is Javascript message~!}&l…

荟萃分析R Meta-Analyses 3 Effect Sizes

总结 效应量是荟萃分析的基石。为了进行荟萃分析&#xff0c;我们至少需要估计效应大小及其标准误差。 效应大小的标准误差代表研究对效应估计的精确程度。荟萃分析以更高的精度和更高的权重给出效应量&#xff0c;因为它们可以更好地估计真实效应。 我们可以在荟萃分析中使用…

【Golang星辰图】大数据时代的明星工具:深入了解Go语言数据处理和分析库

高效处理数据&#xff0c;驾驭大数据时代&#xff1a;深入学习Go语言数据处理库 前言 在当今互联网时代&#xff0c;数据处理和分析变得越来越重要。随着数据规模的不断增大&#xff0c;如何高效地处理和分析数据成为了许多企业和开发者面临的挑战。为了满足这一需求&#xf…

面试算法-102-LRU 缓存

题目 请你设计并实现一个满足 LRU (最近最少使用) 缓存 约束的数据结构。 实现 LRUCache 类&#xff1a; LRUCache(int capacity) 以 正整数 作为容量 capacity 初始化 LRU 缓存 int get(int key) 如果关键字 key 存在于缓存中&#xff0c;则返回关键字的值&#xff0c;否则返…

JavaScript-Web学习笔记02

三、事件高级 1、注册事件&#xff08;绑定事件&#xff09; 注册事件概述 给元素添加事件&#xff0c;称为注册事件或者绑定事件。 注册时间有两种方式&#xff1a;传统方式和方法监听注册方式 传统注册方式 利用 on 开头的事件 onclick<button onclick "alert(hi~)…

软考高级架构师:MVP 架构概念和例题

作者&#xff1a;明明如月学长&#xff0c; CSDN 博客专家&#xff0c;大厂高级 Java 工程师&#xff0c;《性能优化方法论》作者、《解锁大厂思维&#xff1a;剖析《阿里巴巴Java开发手册》》、《再学经典&#xff1a;《Effective Java》独家解析》专栏作者。 热门文章推荐&am…

Spring - AOP/事务 实现原理

AOP 基本概念 官方文档&#xff1a; Aspect Oriented Programming with Spring Spring AOP supports the following AspectJ pointcut designators (PCD) for use in pointcut expressions: within - limits matching to join points within certain types (simply the exec…