Oceanus实践-从0到1开发PG SQL作业

实时即未来，最近在腾讯云Oceanus进行实时计算服务，分享给大家~

本次实践为随机生成的数据写入到Postgres(PG)目的端。

1. 环境搭建

1.1. 创建Oceanus集群

在Oceanus控制台的【集群管理】->【新建集群】页面创建集群，选择地域、可用区、VPC、日志、存储，设置初始密码等。

若之前未使用过VPC，日志(CLS)，存储(COS)这些组件，需要先进行创建。

VPC及子网需要和下面的PG集群使用同一个，否则需要手动打通（如对等连接）。

创建完后的集群如下：

1.2 创建CDW Postgres集群

在云数据仓库控制台创建Postgre集群，这里为了简单，选择了与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,创建表语句如下：

代码语言：txt复制

   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作业，选择刚刚新建的集群创建作业。然后在作业的【开发调试】->【作业参数】里面添加必要的connector，如jdbc connector。

2.2 创建Source端

此处选择Datagen来随机生成一些数据。

代码语言：txt复制

-- 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 ClickHouse中表信息完全对应。

代码语言：txt复制

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 算子操作

这里大家可以自由发挥，进行运算数据处理。我在此处只做了简单的数据插入，没有进行复杂计算。

代码语言：txt复制

   INSERT INTO pg_sink SELECT f_sequence as id, f_random_str as name FROM random_source;

3. 验证总结

查询pg数据，数据是否插入成功。

代码语言：txt复制

select * from testddb.pg_sink;