某些场景下，需要把当前查询的结果，临时写入到未提前定义的或者随时变更的数据库.
本工具提供了 dynamicSourceQuery 的反向操作, 无需建虚拟表可直接向临时数据源写入数据.
目前写入侧实现了 redis 数据源, 支持 string / hash / set / list / zset 五种结构.
语法
- SQL
dynamicSourceWrite(insert into redis_string(key, value) values ('k1', 'v1'))
on url='redis://127.0.0.1:6379/0' and source_type=redis
- 语法
dynamicSourceWrite(要执行的 insert / replace sql) on source_type=’redis’ and url=’redis://127.0.0.1:6379/0’
- 参数 V:0.3.35
| 参数 | 必填 | 默认值 | 举例 | 说明 |
|---|---|---|---|---|
| source_type | 否 | redis | redis | 目标数据源类型, 目前仅支持 redis |
| url | 否 | - | redis://127.0.0.1:6379/0 | redis 直连地址 |
| dbsource | 否 | - | junit_redis | 已配置的 redis 数据源名字, 与 url 二选一 |
| seconds | 否 | 0 | 100 | 写入后的过期时间(秒), 0 表示不过期; 别名: second, expire, ttl |
url 与 dbsource 至少要指定一个, 优先使用 url.
- 注意
- 出于安全考虑, 与 dynamicSourceQuery 一样, 需要在程序中手动注册此命令才能启用.
- 写入必须指定列名, 不支持
insert into redis_string values (...)这种省略列名的写法. - 底层使用 jedis pipeline 批量提交, 单批默认 1000 条.
redis 结构与列名
写入的”表名”决定使用哪种 redis 结构, 各结构要求的列如下:
| 表名 | redis 结构 | 必填列 | 对应命令 |
|---|---|---|---|
| redis_string | string | key, value | SET |
| redis_hash | hash | key, field, value | HSET |
| redis_set | set | key, member | SADD |
| redis_list | list | key, value | RPUSH |
| redis_zset | zset | key, member, score | ZADD |
可选列 seconds (别名 second / expire / ttl) 用于指定该行对应 key 的过期时间. 取值时优先按列名匹配, 匹配不到再按列在语句中的位置匹配.
insert 与 replace 的区别
insert into: 在已有数据上追加或覆盖, 不删除 key.- string:
SET覆盖 value - hash:
HSET按 field 覆盖, 其它 field 保留 - set:
SADD按 member 幂等追加 - list:
RPUSH真追加, 允许重复元素 - zset:
ZADD按 member 覆盖 score
- string:
replace into: 先DEL整个 key, 再写入, 即整 key 覆盖.
注意: 当 key 在 redis 中已存在但结构与写入命令不匹配时(比如 key 是 string 却执行 HSET), 真实 redis 会返回
WRONGTYPE, 本工具会转成中文”数据类型错误”提示. 此时可改用正确的结构, 或使用replace into先清空再写入.
写入样例
- values 直接写入
-- string
dynamicSourceWrite(insert into redis_string(key, value) values ('k1', 'v1'))
on url='redis://127.0.0.1:6379/0' and source_type=redis
-- hash, 同一 key 的多个 field 写成多行
dynamicSourceWrite(insert into redis_hash(key, field, value) values
('h1', 'f1', 'v1'),
('h1', 'f2', 'v2'))
on url='redis://127.0.0.1:6379/0' and source_type=redis
-- zset
dynamicSourceWrite(insert into redis_zset(key, member, score) values
('z1', 'm1', 1.5),
('z1', 'm2', 2.5))
on url='redis://127.0.0.1:6379/0' and source_type=redis
- 指定过期时间
-- ttl 作为一列
dynamicSourceWrite(insert into redis_string(key, value, seconds) values ('k1', 'v1', 100))
on url='redis://127.0.0.1:6379/0' and source_type=redis
-- 或写在 on 参数上, 对本次写入的所有 key 生效
dynamicSourceWrite(insert into redis_string(key, value) values ('k1', 'v1'))
on url='redis://127.0.0.1:6379/0' and source_type=redis and seconds=100
- insert into select
-- 把 select 结果写入 redis, 支持 json / csv 等内联数据源
dynamicSourceWrite(insert into redis_string(key, value)
select key, value from '[{"key":"k1","value":"v1"}]')
on url='redis://127.0.0.1:6379/0' and source_type=redis
结合流水线写入
配合 sysql 的流水线, 可以把上一条查询的结果直接写入 redis, 使用 @sybn_last_select@table 引用上游结果:
-- 第一步: 查询得到结果集
select id, name from user where id < 10
-- 第二步: 把上一步结果写入 redis (无需自己再 select)
dynamicSourceWrite(insert into redis_hash(key, field, value))
on url='redis://127.0.0.1:6379/0' and source_type=redis
语句中既没有写 values 也没有写 select from 时, 自动使用上游数据流作为数据来源.
