Skip to content

Commit 2e4721b

Browse files
todd5167todd5167
committed
db2 doc
1 parent d4540b0 commit 2e4721b

File tree

4 files changed

+233
-56
lines changed

4 files changed

+233
-56
lines changed

‎docs/plugin/db2Side.md‎

Lines changed: 127 additions & 35 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,7 @@
11

22
## 1.格式：
3+
通过建表语句中的` PERIOD FOR SYSTEM_TIME`将表标识为维表，其中`PRIMARY KEY(keyInfo)`中的keyInfo，表示用来和源表进行关联的字段，
4+
维表JOIN的条件必须与`keyInfo`字段一致。
35
```
46
CREATE TABLE tableName(
57
colName cloType,
@@ -25,55 +27,145 @@
2527

2628
## 3.表结构定义
2729

30+
31+
[维表参数信息](docs/plugin/sideParams.md)
32+
db2独有的参数配置：
33+
2834
|参数名称|含义|
2935
|----|---|
30-
| tableName | db2表名称|
31-
| colName | 列名称|
32-
| colType | 列类型 [colType支持的类型](docs/colType.md)|
33-
| PERIOD FOR SYSTEM_TIME | 关键字表明该定义的表为维表信息|
34-
| PRIMARY KEY(keyInfo) | 维表主键定义;多个列之间用逗号隔开|
36+
| type | 维表类型， db2 |是||
37+
| url | 连接数据库 jdbcUrl |是||
38+
| userName | 连接用户名 |是||
39+
| password | 连接密码|是||
40+
| schema | 表所属scheam|否||
3541

36-
## 4.参数
37-
38-
|参数名称|含义|是否必填|默认值|
39-
|----|---|---|----|
40-
| type | 表明维表的类型 db2 |是||
41-
| url | 连接mysql数据库 jdbcUrl |是||
42-
| userName | db2连接用户名 |是||
43-
| password | db2连接密码|是||
44-
| tableName | db2表名称|是||
45-
| cache | 维表缓存策略(NONE/LRU)|否|NONE|
46-
| partitionedJoin | 是否在維表join之前先根据 設定的key 做一次keyby操作(可以減少维表的数据缓存量)|否|false|
47-
48-
----------
49-
> 缓存策略
50-
* NONE: 不做内存缓存
51-
* LRU:
52-
* cacheSize: 缓存的条目数量
53-
* cacheTTLMs:缓存的过期时间(ms)
54-
55-
56-
## 5.样例
42+
43+
44+
## 4.样例
45+
46+
### ALL全量维表定义
5747
```
58-
create table sideTable(
59-
channel varchar,
60-
xccount int,
61-
PRIMARY KEY(channel),
48+
// 定义全量维表
49+
CREATE TABLE sideTable(
50+
id INT,
51+
name VARCHAR,
52+
PRIMARY KEY(id) ,
6253
PERIOD FOR SYSTEM_TIME
6354
)WITH(
6455
type='db2',
65-
url='jdbc:db2://172.16.8.104:50000/test?charset=utf8',
66-
userName='dtstack',
56+
url='jdbc:db2://172.16.10.251:50000/mqTest',
57+
userName='DB2INST1',
6758
password='abc123',
68-
tableName='sidetest',
59+
tableName='USER_INFO2',
60+
schema = 'DTSTACK'
61+
cache ='ALL',
62+
cacheTTLMs ='60000',
63+
parallelism ='2'
64+
);
65+
66+
```
67+
### LRU异步维表定义
68+
69+
```
70+
CREATE TABLE sideTable(
71+
id INT,
72+
name VARCHAR,
73+
PRIMARY KEY(id) ,
74+
PERIOD FOR SYSTEM_TIME
75+
)WITH(
76+
type='db2',
77+
url='jdbc:db2://172.16.10.251:50000/mqTest',
78+
userName='DB2INST1',
79+
password='abc123',
80+
tableName='USER_INFO2',
81+
schema = 'DTSTACK'
82+
partitionedJoin ='false',
6983
cache ='LRU',
7084
cacheSize ='10000',
7185
cacheTTLMs ='60000',
72-
parallelism ='1',
73-
partitionedJoin='false'
86+
asyncPoolSize ='3',
87+
parallelism ='2'
88+
);
89+
90+
```
91+
92+
93+
### DB2异步维表关联输出到Console
94+
```
95+
96+
CREATE TABLE source1 (
97+
id int,
98+
name VARCHAR
99+
)WITH(
100+
type ='kafka11',
101+
bootstrapServers ='172.16.8.107:9092',
102+
zookeeperQuorum ='172.16.8.107:2181/kafka',
103+
offsetReset ='latest',
104+
topic ='mqTest03',
105+
timezone='Asia/Shanghai',
106+
topicIsPattern ='false'
74107
);
75108
76109
110+
111+
CREATE TABLE source2(
112+
id int,
113+
address VARCHAR,
114+
PERIOD FOR SYSTEM_TIME
115+
)WITH(
116+
type='db2',
117+
url='jdbc:db2://172.16.10.251:50000/mqTest',
118+
userName='DB2INST1',
119+
password='abc123',
120+
tableName='USER_INFO2',
121+
schema = 'DTSTACK',
122+
batchSize = '1'
123+
);
124+
125+
126+
CREATE TABLE MyResult(
127+
id int,
128+
name VARCHAR,
129+
address VARCHAR,
130+
primary key (id)
131+
)WITH(
132+
type='console'
133+
);
134+
135+
insert into MyResult
136+
select
137+
s1.id,
138+
s1.name,
139+
s2.address
140+
from
141+
source1 s1
142+
left join
143+
source2 s2
144+
on
145+
s1.id = s2.id
146+
77147
```
78148

149+
DB2维表字段信息
150+
151+
```aidl
152+
-- DTSTACK.USER_INFO2 definition
153+
154+
CREATE TABLE "DTSTACK "."USER_INFO2" (
155+
"ID" INTEGER ,
156+
"NAME" VARCHAR(50 OCTETS) ,
157+
"ADDRESS" VARCHAR(50 OCTETS) )
158+
IN "USERSPACE1"
159+
ORGANIZE BY ROW
160+
;
161+
162+
GRANT CONTROL ON TABLE "DTSTACK "."USER_INFO2" TO USER "DB2INST1"
163+
;
164+
```
165+
166+
维表数据：(1001,maqi,hz)
167+
168+
源表数据：{"name":"maqi","id":1001}
169+
79170

171+
输出结果： (1001,maqi,hz)

‎docs/plugin/db2Sink.md‎

Lines changed: 102 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,8 @@
33
CREATE TABLE tableName(
44
colName colType,
55
...
6-
colNameX colType
6+
colNameX colType,
7+
[primary key (colName)]
78
)WITH(
89
type ='db2',
910
url ='jdbcUrl',
@@ -25,29 +26,112 @@ CREATE TABLE tableName(
2526
| tableName| db2表名称|
2627
| colName | 列名称|
2728
| colType | 列类型 [colType支持的类型](docs/colType.md)|
29+
| primary key | updateMode为UPSERT时，需要指定的主键信息，不需要和数据库一致|
30+
2831

2932
## 4.参数：
3033

3134
|参数名称|含义|是否必填|默认值|
3235
|----|----|----|----|
33-
| type |表名 输出表类型[mysq|hbase|elasticsearch]|是||
34-
| url | 连接db2数据库 jdbcUrl |是||
35-
| userName | db2连接用户名 |是||
36-
| password | db2连接密码|是||
37-
| tableName | db2表名称|是||
38-
| parallelism | 并行度设置|否|1|
36+
|type |结果表插件类型，必须为db2|是||
37+
|url | 连接db2数据库 jdbcUrl |是||
38+
|userName |db2连接用户名 |是||
39+
|password | db2连接密码|是||
40+
|tableName | db2表名称|是||
41+
|schema | db2表空间|否||
42+
|parallelism | 并行度设置|否|1|
43+
|batchSize | flush的大小|否|100|
44+
|batchWaitInterval | flush的时间间隔，单位ms|否|1000|
45+
|allReplace| true:新值替换旧值|否|false|
46+
|updateMode| APPEND：不回撤数据，只下发增量数据，UPSERT：先删除回撤数据，然后更新|否|结果表设置主键则为UPSERT|
3947

4048
## 5.样例：
49+
50+
回溯流删除
51+
4152
```
53+
54+
CREATE TABLE source1 (
55+
id int,
56+
name VARCHAR
57+
)WITH(
58+
type ='kafka11',
59+
bootstrapServers ='172.16.8.107:9092',
60+
zookeeperQuorum ='172.16.8.107:2181/kafka',
61+
offsetReset ='latest',
62+
topic ='mqTest03',
63+
timezone='Asia/Shanghai',
64+
topicIsPattern ='false'
65+
);
66+
67+
68+
69+
CREATE TABLE source2(
70+
id int,
71+
address VARCHAR
72+
)WITH(
73+
type ='kafka11',
74+
bootstrapServers ='172.16.8.107:9092',
75+
zookeeperQuorum ='172.16.8.107:2181/kafka',
76+
offsetReset ='latest',
77+
topic ='mqTest04',
78+
timezone='Asia/Shanghai',
79+
topicIsPattern ='false'
80+
);
81+
82+
4283
CREATE TABLE MyResult(
43-
channel VARCHAR,
44-
pv VARCHAR
45-
)WITH(
46-
type ='db2',
47-
url ='jdbc:db2://172.16.8.104:50000/test?charset=utf8',
48-
userName ='dtstack',
49-
password ='abc123',
50-
tableName ='pv2',
51-
parallelism ='1'
52-
)
53-
```
84+
id int,
85+
name VARCHAR,
86+
address VARCHAR,
87+
primary key (id)
88+
)WITH(
89+
type='db2',
90+
url='jdbc:db2://172.16.10.251:50000/mqTest',
91+
userName='DB2INST1',
92+
password='abc123',
93+
tableName='USER_INFO2',
94+
schema = 'DTSTACK',
95+
updateMode = 'upsert', // 设置 primary key则默认为upsert
96+
batchSize = '1'
97+
);
98+
99+
insert into MyResult
100+
select
101+
s1.id,
102+
s1.name,
103+
s2.address
104+
from
105+
source1 s1
106+
left join
107+
source2 s2
108+
on
109+
s1.id = s2.id
110+
111+
112+
113+
```
114+
115+
116+
DB2结果表建表语句：
117+
118+
```aidl
119+
CREATE TABLE "DTSTACK "."USER_INFO2" (
120+
"ID" INTEGER ,
121+
"NAME" VARCHAR(50 OCTETS) ,
122+
"ADDRESS" VARCHAR(50 OCTETS) )
123+
IN "USERSPACE1"
124+
ORGANIZE BY ROW
125+
;
126+
127+
GRANT CONTROL ON TABLE "DTSTACK "."USER_INFO2" TO USER "DB2INST1"
128+
;
129+
```
130+
131+
132+
133+
数据结果：
134+
135+
向Topic mqTest03 发送数据 {"name":"maqi","id":1001} 插入 (1001,"maqi",null)
136+
137+
向Topic mqTest04 发送数据 {"address":"hz","id":1001} 删除 (1001,"maqi",null) 插入 (1001,"maqi","hz")

‎docs/plugin/mysqlSink.md‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,8 @@
33
CREATE TABLE tableName(
44
colName colType,
55
...
6-
colNameX colType
6+
colNameX colType,
7+
[primary key (colName)]
78
)WITH(
89
type ='mysql',
910
url ='jdbcUrl',
@@ -39,7 +40,7 @@ CREATE TABLE tableName(
3940
|batchSize | flush的大小|否|100|
4041
|batchWaitInterval | flush的时间间隔，单位ms|否|1000|
4142
|allReplace| true:新值替换旧值|否|false|
42-
|updateMode| APPEND：不回撤数据，只下发增量数据，UPSERT：先删除回撤数据，然后更新|否||
43+
|updateMode| APPEND：不回撤数据，只下发增量数据，UPSERT：先删除回撤数据，然后更新|否|结果表设置主键则为UPSERT|
4344

4445
## 5.完整样例：
4546
```

‎docs/plugin/postgresqlSink.md‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -43,7 +43,7 @@ CREATE TABLE tableName(
4343
|batchSize | flush的大小|否|100|
4444
|batchWaitInterval | flush的时间间隔，单位ms|否|1000|
4545
|allReplace| true:新值替换旧值|否|false|
46-
|updateMode| APPEND：不回撤数据，只下发增量数据，UPSERT：先删除回撤数据，然后更新|否||
46+
|updateMode| APPEND：不回撤数据，只下发增量数据，UPSERT：先删除回撤数据，然后更新|否|结果表设置主键则为UPSERT|
4747

4848
## 5.样例：
4949

0 commit comments

Comments
 (0)