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

流计算 Oceanus 简介

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

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

前置准备

创建流计算 Oceanus 集群

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

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

创建 PostgreSQL 实例

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

代码语言：sql复制

-- 用于 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 字段。