Flink|Flink 实践教程(入门(6):读取 PG 数据写入 ClickHouse)
作者:腾讯云流计算 Oceanus 团队
流计算 Oceanus 简介
流计算 Oceanus 是大数据产品生态体系的实时化分析利器,是基于 Apache Flink 构建的具备一站开发、无缝连接、亚秒延时、低廉成本、安全稳定等特点的企业级实时大数据分析平台。流计算 Oceanus 以实现企业数据价值最大化为目标,加速企业实时化数字化的建设进程。
本文将向您详细介绍如何获取 PostgreSQL 表数据,并使用字符串函数进行转换,最后将数据输出到 ClickHouse 中。
操作视频
前置准备
创建流计算 Oceanus 集群
进入流计算 Oceanus 控制台 [1],点击左侧【集群管理】,点击左上方【创建集群】,具体可参考流计算 Oceanus 官方文档 创建独享集群 [2]。
创建 PostgreSQL 实例
进入 PostgreSQL 控制台 [3],点击左上角【新建】创建实例,具体参考 创建 PostgreSQL 实例 [4]。
数据准备:
进入实例数据库,创建 test1 表,并手动插入数据。
-- 建表语句
create table public.test1 (
id INT,
str_one VARCHAR(50),
str_two VARCHAR(50),
str_thr VARCHAR(50),
PRIMARY key(id)
);
?
-- 插入语句
INSERT INTO public.test1 VALUES (1, 'hello world', 'b', 'Oceanus-1');
INSERT INTO public.test1 VALUES (2, 'good job', 'c', 'Oceanus-2');
INSERT INTO public.test1 VALUES (3, 'hello oceanus', 'd', 'Oceanus-3');
笔者这里使用 DBeaver 进行外网连接,更多连接方式参考官网文档 连接 PostgreSQL 实例 [5]创建 ClickHouse 集群
进入 ClickHouse 控制台 [6],点击左上角【新建集群】,完成 ClickHouse 集群创建,具体可参考 ClickHouse 快速入门 [7]。创建 ClickHouse 表:登陆 ClickHouse 集群(登入方式参考 ClickHouse 快速入门 [7]),并建表。
CREATE TABLE default.pg_to_ck on cluster default_cluster (
idInt8,
str_one String,
str_two String,
str_thr String,
Sign Int8 )
ENGINE = ReplicatedCollapsingMergeTree('/clickhouse/tables/{layer}-{shard}/default/pg_to_ck', '{replica}',Sign)
ORDER BY (id);
注:流计算 Oceanus 集群、PostgreSQL 实例、ClickHouse 集群需在同一 VPC 下。流计算 Oceanus 作业 1. 创建 Source
-- PostgreSQL CDC Source。
CREATE TABLE PostgreSourceTable (
id INT,
str_one VARCHAR,
str_two VARCHAR,
str_thr VARCHAR,
PRIMARY KEY (id) NOT ENFORCED -- 如果要同步的数据库表定义了主键, 则这里也需要定义
) WITH (
'connector' = 'postgres-cdc',-- 必须为 'postgres-cdc'
'hostname' = '10.0.0.236',-- 数据库的 IP
'port' = '5432',-- 数据库的访问端口
'username' = 'root',-- 数据库访问使用的用户名(需要提供 REPLICATION 权限, 日志级别必须大于等于 logical, 且设置后需要重启实例)
'password' = 'xxxxxxxxxxx',-- 数据库访问使用的密码
'database-name' = 'postgres',-- 需要同步的数据库名
'schema-name' = 'public',-- 需要同步的数据库模式 (Schema)
'table-name' = 'test1'-- 需要同步的数据表名
);
2. 创建 Sink
-- ClickHouse Sink (不完全支持upsert,详见说明文档)。配合 flink-connector-clickhouse 使用。
CREATE TABLE clickhouse_sink (
id INT,
str_one VARCHAR,
str_two VARCHAR,
str_thr VARCHAR,
PRIMARY KEY (id) NOT ENFORCED -- 如果要同步的数据库表定义了主键, 则这里也需要定义
) WITH (
'connector' = 'clickhouse',-- connector 类型为 clickhouse
'url' = 'clickhouse://10.0.0.178:8123',-- 指定数据库链接 url
'database-name' = 'default',-- 需要写入的 clickhouse 库名
'table-name' = 'pg_to_ck',-- 需要写入的 clickhouse 表名
'table.collapsing.field' = 'Sign'-- 采用 CollapsingMergeTree 引擎的 clickhouse 表,Collapsing 列字段的名称
);
3. 编写业务 SQL
INSERT INTO clickhouse_sink
SELECT
id,
--INITCAP:将 str_one 中的单词转为大写开头,例如 INITCAP('i have a dream') 返回 'I Have A Dream'。
INITCAP(str_one)AS str_one,
--TO_BASE64:将 string 表示的字符串编码为 Base64 字符串。
TO_BASE64(str_two)AS str_two,
--REPLACE:将 string1 字符串中所有的 string2 替换为 string3。例如 REPLACE('banana', 'a', 'A') 返回 'bAnAnA'。
REPLACE(str_thr,'Oceanus','Hello Oceanus') AS str_thr
FROM PostgreSourceTable;
这里我们使用 Flink 1.13 集群,旧版 Flink 集群需选择相应的内置 Connector总结
- 使用 Postgres-CDC 连接器:
- 用于同步的 Postgres 用户至少需要开启 REPLICATION、LOGIN、SCHEMA、DATABASE、SELECT 权限。可以进入 PostgreSQL 数据库进行授权操作。
CREATE ROLE debezium_user REPLICATION LOGIN;
GRANT USAGE ON DATABASE database_name TO debezium_user;
GRANT USAGE ON SCHEMA schema_name TO debezium_user;
GRANT SELECT ON scheam_name.table_name, scheam_name.table_name TO debezium_user;
- 日志级别必须大于等于 logical, 且设置后需要重启实例。进入数据库实例,单击【参数设置】,单击【WAL】,修改【wal_level】的【参数运行值】为 "logical"。修改成功后点击右上角【重启】。
- 更多字符串操作函数请参考流计算 Oceanus 官方文档 字符串函数[8]。
[2] 创建独享集群:https://cloud.tencent.com/document/product/849/48298
[3] PostgreSQL 控制台:https://console.cloud.tencent.com/postgres/index
[4] 创建 PostgreSQL 实例:https://cloud.tencent.com/document/product/409/56961
[5] 连接 PostgreSQL 实例:https://cloud.tencent.com/document/product/409/40429
[6] ClickHouse 控制台:https://console.cloud.tencent.com/cdwch?region=ap-guangzhou
[7] ClickHouse 快速入门:https://cloud.tencent.com/document/product/1299/49824
[8] 流计算 Oceanus 字符串函数:https://cloud.tencent.com/document/product/849/18073
文章图片
关注“腾讯云大数据”公众号,技术交流、最新活动、服务专享一站 Get~
流计算 Oceanus 限量秒杀专享活动火爆进行中↓↓
文章图片
【Flink|Flink 实践教程(入门(6):读取 PG 数据写入 ClickHouse)】
推荐阅读
- 2.6|2.6 Photoshop操作步骤的撤消和重做 [Ps教程]
- 漫画初学者如何学习漫画背景的透视画法(这篇教程请收藏好了!)
- 不废话,代码实践带你掌握|不废话,代码实践带你掌握 强缓存、协商缓存!
- 六项精进20180530
- 数据库|SQL行转列方式优化查询性能实践
- 【Day31课后实践】
- 用npm发布一个包的教程并编写一个vue的插件发布
- 20180322【w4复盘日志】
- 狗狗定点大小便视频教程下载地址
- SwiftUI|SwiftUI iOS 瀑布流组件之仿CollectionView不规则图文混合(教程含源码)