Skip to content

Commit 93844cf

Browse files
todd5167todd5167
committed
pgsink sqlserversink doc
1 parent b231974 commit 93844cf

3 files changed

Lines changed: 164 additions & 35 deletions

File tree

‎docs/plugin/mysqlSink.md‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -36,7 +36,7 @@ CREATE TABLE tableName(
3636
|password | mysql连接密码|是||
3737
|tableName | mysql表名称|是||
3838
|parallelism | 并行度设置|否|1|
39-
|batchNum | flush的大小|否|100|
39+
|batchSize | flush的大小|否|100|
4040
|batchWaitInterval | flush的时间间隔，单位ms|否|1000|
4141
|allReplace| true:新值替换旧值|否|false|
4242
|updateMode| APPEND：不回撤数据，只下发增量数据，UPSERT：先删除回撤数据，然后更新|否||

‎docs/plugin/postgresqlSink.md‎

Lines changed: 84 additions & 17 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 ='postgresql',
910
url ='jdbcUrl',
@@ -25,29 +26,95 @@ CREATE TABLE tableName(
2526
| tableName| 在 sql 中使用的名称;即注册到flink-table-env上的名称|
2627
| colName | 列名称|
2728
| colType | 列类型 [colType支持的类型](docs/colType.md)|
29+
| primary key | updateMode为UPSERT时，需要指定的主键信息|
30+
2831

2932
## 4.参数：
3033

3134
|参数名称|含义|是否必填|默认值|
3235
|----|----|----|----|
33-
| type |表明 输出表类型[postgresql]|是||
34-
| url | 连接postgresql数据库 jdbcUrl |是||
35-
| userName | postgresql连接用户名 |是||
36-
| password | postgresql连接密码|是||
37-
| tableName | postgresqll表名称|是||
38-
| parallelism | 并行度设置|否|1|
36+
|type |结果表插件类型，必须为postgresql|是||
37+
|url | 连接postgresql数据库 jdbcUrl |是||
38+
|userName |postgresql连接用户名 |是||
39+
|password | postgresql连接密码|是||
40+
|tableName | postgresql表名称|是||
41+
|schema | postgresql表空间|否||
42+
|parallelism | 并行度设置|否|1|
43+
|batchSize | flush的大小|否|100|
44+
|batchWaitInterval | flush的时间间隔，单位ms|否|1000|
45+
|allReplace| true:新值替换旧值|否|false|
46+
|updateMode| APPEND：不回撤数据，只下发增量数据，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 ='postgresql',
47-
url ='jdbc:postgresql://localhost:9001/test?sslmode=disable',
48-
userName ='dtstack',
49-
password ='abc123',
50-
tableName ='pv2',
51-
parallelism ='1'
52-
)
84+
id int,
85+
name VARCHAR,
86+
address VARCHAR,
87+
primary key (id)
88+
)WITH(
89+
type='postgresql',
90+
url='jdbc:postgresql://172.16.8.193:5432/DTstack',
91+
userName='root',
92+
password='123456',
93+
tableName='userInfo',
94+
schema = 'aaa',
95+
updateMode = '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+
53113
```
114+
115+
116+
数据结果：
117+
118+
向Topic mqTest03 发送数据 {"name":"maqi","id":1001} 插入 (1001,"maqi",null)
119+
120+
向Topic mqTest04 发送数据 {"address":"hz","id":1001} 删除 (1001,"maqi",null) 插入 (1001,"maqi","hz")

‎docs/plugin/sqlserverSink.md‎

Lines changed: 79 additions & 17 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 ='sqlserver',
910
url ='jdbcUrl',
@@ -25,29 +26,90 @@ CREATE TABLE tableName(
2526
| tableName| sqlserver表名称|
2627
| colName | 列名称|
2728
| colType | 列类型 [colType支持的类型](docs/colType.md)|
29+
| primary key | updateMode为UPSERT时，需要指定的主键信息|
2830

2931
## 4.参数：
3032

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

4048
## 5.样例：
49+
50+
回溯流删除
4151
```
52+
53+
CREATE TABLE source1 (
54+
id int,
55+
name VARCHAR
56+
)WITH(
57+
type ='kafka11',
58+
bootstrapServers ='172.16.8.107:9092',
59+
zookeeperQuorum ='172.16.8.107:2181/kafka',
60+
offsetReset ='latest',
61+
topic ='mqTest03',
62+
timezone='Asia/Shanghai',
63+
topicIsPattern ='false'
64+
);
65+
66+
67+
CREATE TABLE source2(
68+
id int,
69+
address VARCHAR
70+
)WITH(
71+
type ='kafka11',
72+
bootstrapServers ='172.16.8.107:9092',
73+
zookeeperQuorum ='172.16.8.107:2181/kafka',
74+
offsetReset ='latest',
75+
topic ='mqTest04',
76+
timezone='Asia/Shanghai',
77+
topicIsPattern ='false'
78+
);
79+
80+
4281
CREATE TABLE MyResult(
43-
channel VARCHAR,
44-
pv VARCHAR
45-
)WITH(
46-
type ='sqlserver',
47-
url ='jdbc:jtds:sqlserver://172.16.8.104:1433;DatabaseName=mytest',
48-
userName ='dtstack',
49-
password ='abc123',
50-
tableName ='pv2',
51-
parallelism ='1'
52-
)
53-
```
82+
id int,
83+
name VARCHAR,
84+
address VARCHAR,
85+
primary key (id)
86+
)WITH(
87+
type='sqlserver',
88+
url='jdbc:jtds:sqlserver://172.16.8.149:1433;DatabaseName=DTstack',
89+
userName='sa',
90+
password='Dtstack2018',
91+
tableName='user',
92+
schema = 'aaa',
93+
updateMode = 'upsert',
94+
batchSize = '1'
95+
);
96+
97+
insert into MyResult
98+
select
99+
s1.id,
100+
s1.name,
101+
s2.address
102+
from
103+
source1 s1
104+
left join
105+
source2 s2
106+
on
107+
s1.id = s2.id
108+
109+
```
110+
111+
数据结果：
112+
113+
向Topic mqTest03 发送数据 {"name":"maqi","id":1001} 插入 (1001,"maqi",null)
114+
115+
向Topic mqTest04 发送数据 {"address":"hz","id":1001} 删除 (1001,"maqi",null) 插入 (1001,"maqi","hz")

0 commit comments

Comments
 (0)