1. 极安网首页
  2. 网络安全新闻

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

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

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

1. 环境搭建

1.1. 创建Oceanus集群

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

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

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

创建完后的集群如下:

Oceanus集群

1.2 创建CDW Postgres集群

在云数据仓库控制台创建Postgre集群,这里为了简单,选择了与Oceanus同一个地域,同可用区。网络选择也选择与上面同样的VPC。

创建完后的集群如下:

CDW Postgres集群

要登录集群目前需要在同网段的云服务器才能登录,集群节点不能直接登录,通过同网段的云服务器(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作业,选择刚刚新建的集群创建作业。然后在作业的【开发调试】->【作业参数】里面添加必要的connector,如jdbc 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 ClickHouse中表信息完全对应。

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;

相关链接:

Oceanus官方文档:https://cloud.tencent.com/document/product/849

PG集群官方:https://cloud.tencent.com/document/product/878

文章来源于互联网:Oceanus实践-从0到1开发PG SQL作业

原创文章,作者: Admin ,转载请注明出处:https://secvery.com/9063.html