Oceanus 实践-从0到1开发PG SQL作业
实时即未来，最近在腾讯云流计算 Oceanus 进行 Flink 实时计算服务，分享给大家~
本次实践为随机生成的数据写入到 Postgres(PG) 目的端。
1. 环境搭建
1.1. 创建流计算 Oceanus 集群
在 Oceanus 控制台的【集群管理】->【新建集群】页面创建集群，选择地域、可用区、VPC、日志、存储，设置初始密码等。
若之前未使用过 VPC，日志(CLS)，存储(COS)这些组件，需要先进行创建。
VPC及子网需要和下面的PG集群使用同一个，否则需要手动打通（如对等连接）。
创建完后的集群如下：
1.2 创建 CDW PostgreSQL 集群
在云数据仓库控制台创建 PostgreSQL 集群，这里为了简单，选择了与 Oceanus 同一个地域，同可用区。网络选择也选择与上面同样的 VPC。
创建完后的集群如下：
要登录集群目前需要在同网段的云服务器才能登录，集群节点不能直接登录，通过同网段的云服务器（CVM)安装pg客户端进行登录，创建 PG 表。
这里需要先登录postgres database，然后创建自己的database。创建后进行才能登录自己的数据库，登录命令：
psql -h 172.28.28.91 -p 5436 -U test_root -d postgres
create database testdb;
\q
psql -h 172.28.28.91 -p 5436 -U test_root -d postgres
创建schema,创建表语句如下：
create schema testschema;
create table testschema.pg_sink (id int not null, name text, primary key(id)) distributed by (id);至此，环境准备完毕。
2. 作业创建
2.1 创建 SQL 作业
在 Oceanus 控制台【作业管理】->【新建作业】-> SQL作业，选择刚刚新建的集群创建作业。1.11 以下版本需在作业的【开发调试】->【作业参数】里面添加必要的 connector，如 jdbc connector。当前版本兼容了 1.13 Flink 无需手动添加 connector。
2.2 创建Source端
此处选择 Datagen 来随机生成一些数据。
-- Datagen Connector 可以随机生成一些数据用于测试
-- 参见 [https://ci.apache.org/projects/flink/flink-docs-release-1.11/zh/dev/table/connectors/datagen.html](https://ci.apache.org/projects/flink/flink-docs-release-1.11/zh/dev/table/connectors/datagen.html)
CREATE TABLE random_source (
f_sequence INT,
f_random INT,
f_random_str VARCHAR
) WITH (
'connector' = 'datagen',
'rows-per-second'='1', -- 每秒产生的数据条数
'fields.f_sequence.kind'='sequence', -- 有界序列（结束后自动停止输出）
'fields.f_sequence.start'='1', -- 序列的起始值
'fields.f_sequence.end'='10000', -- 序列的终止值
'fields.f_random.kind'='random', -- 无界的随机数
'fields.f_random.min'='1', -- 随机数的最小值
'fields.f_random.max'='1000', -- 随机数的最大值
'fields.f_random_str.length'='10' -- 随机字符串的长度
);2.3 创建 Sink 端
此处 sink 的表信息需要和 CDW PG 中表信息完全对应。
CREATE TABLE `pg_sink` (
`id` INT,
`name` VARCHAR
) WITH (
-- 指定数据库连接参数
'connector' = 'jdbc',
'url' = 'jdbc:postgresql://172.28.28.91:5436/testdb?currentSchema=testschema&reWriteBatchedInserts=true', -- 请替换为您的实际 PostgreSQL 连接参数
'table-name' = 'pg_sink', -- 需要写入的数据表
'username' = 'test_root', -- 数据库访问的用户名（需要提供 INSERT 权限）
'password' = 'your_password', -- 数据库访问的密码
-- 数据目的 Sink 性能调优参数
'sink.buffer-flush.max-rows' = '5000', -- 可选参数, 表示每批数据的最大缓存条数, 默认值是 5000
'sink.buffer-flush.interval' = '2s', -- 可选参数, 表示每批数据的刷新周期, 默认值是 0s
'sink.max-retries' = '3' -- 可选参数, 表示数据库写入出错时, 最多尝试的次数
);2.4 算子操作
这里大家可以自由发挥，进行运算数据处理。我在此处只做了简单的数据插入，没有进行复杂计算。
INSERT INTO pg_sink SELECT f_sequence as id, f_random_str as name FROM random_source;3. 验证总结
查询pg数据，数据是否插入成功。
select * from testddb.pg_sink;阅读参考：
[1] 流计算 Oceanus 官方文档：https://cloud.tencent.com/document/product/849
[2] PostgreSQL (CDW PG) 集群官方文档：https://cloud.tencent.com/document/product/878
腾讯云开发者
