Flink 实践教程：入门8-简单 ETL 作业

原创

吴云涛

修改于 2021-12-08 15:59:51

9403

修改于 2021-12-08 15:59:51

流计算 Oceanus 简介

流计算 Oceanus 是大数据产品生态体系的实时化分析利器，是基于 Apache Flink 构建的具备一站开发、无缝连接、亚秒延时、低廉成本、安全稳定等特点的企业级实时大数据分析平台。流计算 Oceanus 以实现企业数据价值最大化为目标，加速企业实时化数字化的建设进程。

本示例使用 Oceanus 平台的 ETL 功能，将 PostgreSQL 数据取出，经过时间转换函数处理后存入 PostgreSQL 中。用户无需编写 SQL 代码，只用在界面上进行简单的点击操作即可创建 Oceanus ETL 作业。

Flink 实践教程：入门8-简单 ETL 作业

前置准备

创建流计算 Oceanus 集群

活动购买链接 1 元购买 Oceanus 集群。

进入 Oceanus 控制台 [1]，点击左侧【集群管理】，点击左上方【创建集群】，具体可参考 Oceanus 官方文档 创建独享集群 [2]。

创建 PostgreSQL 实例

进入 PostgreSQL 控制台 [3]，点击左上角【新建】创建实例，具体参考 创建 PostgreSQL 实例 [4]。进入实例数据库，创建表。

-- 用于 Source
CREATE TABLE public.oceanus8_output (
id         INT,
time_one   TIMESTAMP,
PRIMARY    KEY(id)   );

-- 手动插入数据
INSERT INTO public.oceanus8_output VALUES (1,'2020-10-01 18:00:00');
INSERT INTO public.oceanus8_output VALUES (2,'2021-10-01 18:30:24');

-- 用于 Sink
CREATE TABLE public.oceanus8_input (
id           INT,
transf_one   VARCHAR(50),
transf_two   TIMESTAMP,
const_four   INT,
PRIMARY      KEY(id)  );

笔者这里使用 DBeaver 进行外网连接，更多连接方式参考官网文档 连接 PostgreSQL 实例 [5]

流计算 Oceanus 作业

进入 Oceanus 控制台 [1]，点击左上角【新建】创建 ETL 作业，点击【开发调试】进入作业编辑页面。ETL 作业源端可以是 MySQL、PostgreSQL，目的端可以是 MySQL、PostgreSQL、ClickHouse 和 Elasticsearch。

1. 创建 Source

单击【数据源表】右侧【添加】按钮，选择 PostgreSQL ，选择并填写数据库表相关的信息。

2. 创建 Sink

单击【数据目的表】右侧【添加】按钮，选择 PostgreSQL ，选择并填写数据库表相关的信息。

3. 映射字段

编写需要创建映射字段的业务逻辑。这里使用 DATA_FORMAT 函数将 time_one 字段类型由 TIMESTAMP 映射为 STRING，使用 TIMESTAMPADD 函数将 time_one 字段增加一周，并将常量 1000 存入 const_str 字段。