最近更新时间:2023.08.18 17:27:51
首次发布时间:2022.08.09 14:26:49
DataLeap接入了流式计算 Flink 版,在关联 Flink 的项目和资源池后,可以进行 Flink 作业开发。可以通过 Serverless Flink SQL 作业实现不同存储系统之间的 ETL 等。本文以一个简单的示例,将为您介绍 Serverless Flink SQL作业相关的开发流程操作。
DataLeap产品需开通 DataOps敏捷研发、大数据分析、数据开发特惠版或分布式数据自治服务后,才可绑定流式计算 Flink 引擎。绑定引擎操作详见:项目管理。
子账号操作项目绑定 Flink 引擎实例时:
主账号需要先在流式计算 Flink 版控制台导入 IAM 用户。操作详见:用户管理。
并将子账号加入到对应引擎项目中。操作详见:引擎项目成员管理。
目前 Flink 版本只支持 1.11。
在概览界面,显示加入的项目中,单击数据开发进入对应项目。
在任务开发界面,左侧导航栏中,单击新建任务按钮,进入新建任务页面。
选择任务类型:
分类:数据开发。
绑定引擎:流式计算 Flink 版。
关联引擎项目:默认选择引擎绑定时选择的引擎项目,不可更改。
选择任务:流式数据 Serverless Flink SQL。
填写任务基本信息:
任务名称:输入任务的名称,只能由数字、字母、下划线、-和.组成, 首尾只能是数字、字母,且允许输入 1~63 个字符。
保存至: 选择任务存放的目标文件夹目录。
单击确认按钮,成功创建任务。
新建任务成功后,进入代码开发编辑界面,通过 DDL 和 DML 编辑 SQL ,示例代码如下:
详细语法可参考Flink 1.11 官方文档。
以下为示例Demo:将 Datagen 连接器生成的随机数,输出并打印到 Stdout(标准输出)日志中。
CREATE TABLE datagen_source ( siteid INT, citycode SMALLINT, username STRING, pv BIGINT ) WITH ( 'connector' = 'datagen', 'rows-per-second' = '5', 'fields.username.length' = '30', 'fields.siteid.max' = '1000000000', 'fields.siteid.min' = '10000' ); CREATE TABLE print_sink ( siteid INT, citycode SMALLINT, username STRING, pv BIGINT ) WITH ( 'connector' = 'print', 'print-identifier' = 'out' ); insert into print_sink select * from datagen_source;
SQL 编辑完成后,您可在下方选择数据源格式:
支持选择其他和 Pb 数据类型。
Pb 类定义:需要将 Pb 类定义文件拖动到输入框中,或手动输入,如以下示例:
一次只能有一个入口类。
若入口类下的字段有嵌套其他类,这些类必须放在一起定义。
填写示例: syntax = "proto2"; package abase_test; message AbaseTest { required int64 first_id = 1; required int64 latest_id = 2; }
Pb 入口 message:填写类名信息
功能名称 | 描述 |
---|---|
格式化 | 依据在个性化设置中的SQL格式化风格的设置,格式化书写的代码,使其语法结构看起来简洁明了。 |
解析 | 解析检查书写的 SQL 代码的语法和语义正确性,运行前检查语法错误信息,防止运行出错。 |
执行引擎 | 目前只支持 Flink 1.11。 |
单击右侧导航栏中参数设置,进行任务的基本信息、任务输入参数、资源设置、数据源登记、Flink 运行参数配置。
Serverless Flink SQL 任务的基本信息配置如下:
参数名称 | 描述 |
---|---|
任务名称 | 显示创建任务时输入的任务名称,参数设置中不支持修改,可以在左侧任务目录结构中的任务名称右侧更多单击重命名进行修改。 |
任务类型 | Serverless Flink SQL |
引擎类型 | 流式计算 Flink 版。 |
关联引擎项目 | DataLeap侧关联的引擎项目名称。 |
任务描述 | 非必填,可对任务进行详细描述,方便后续查看和管理。 |
责任人 | 仅限一个成员,默认为任务创建人(任务执行失败、复查通过或者失败时的默认接收者),可根据实际需要,修改为其他项目成员。
|
计算资源 | 从关联的引擎项目中选择目标资源池。 |
标签 | 您可以自定义标签,用于标识某一类任务,以便快速搜索过滤,操作即时生效,无需重新上线任务。
|
您可通过手动添加的方式,选择项目参数或自定义参数,以变量或常量形式传入流式任务中使用,来实现同一套代码在不同执行环境下能够自动替换参数来执行。添加操作详见 3.5.2-任务输入参数。
设置任务运行时相关资源分配情况:
参数名称 | 描述 |
---|---|
TaskManager个数 | 设置 flink 作业中 TaskManager 的数量。 |
单TaskManagerCPU数 | 设置单个 TaskManager 所占用的CPU数量。 |
单TaskManager内存大小(MB) | 设置单个 TaskManager 所占用的内存大小。 |
单TaskManager slot数 | 设置单个 TaskManager 中slot的数量。 |
JobManager CPU数 | 设置单个 JobManager 所占用的CPU数量。 |
JobManager内存 | 设置单个 JobManager 所占用的内存大小。 |
登记该任务使用的 Source 和 Sink 信息,以用于后续监控配置和血缘构建。支持通过自动解析和手动添加方式。
自动解析:保存 SQL 逻辑后,单击自动解析按钮,可根据 SQL 逻辑自动进行数据源上下游登记。
源/目标:支持源/目标 Source 和 Sink 的选择。
数据源类型:支持选择 DataGen、Rmq、Bmq、JDBC、Abase、火山/自建 Kafka 这些 Connector 的数据源自动解析。
说明
Abase 和 Rmq 数据源类型,仅支持解析操作,暂不支持上线发布后运行。且 Abase 数据源通常是在火山引擎内部业务上云中使用。
各数据源类型需补充填写相应的信息,您可根据实际场景进行配置,如 Kafka 的数据源类型,需填写具体的Bootstrap Servers、Topic 名称、消费组等信息。
直接上游任务:根据数据源匹配逻辑,自动解析出上游任务。您也可通过下方手动添加的方式,进行搜索上游任务名称,单击添加按钮,进行手动添加。
直接下游任务:根据数据源匹配逻辑,自动解析出下游任务,但无法通过手动添加的方式进行操作,仅展现通过数据源匹配后的结果。
自动解析的数据源,您可通过关闭自动解析按钮,将登记的数据源给删除。
手动添加:您也可以通过手动添加的方式进行数据源登记。
批量删除:手动添加的数据源,您可通过勾选已登记的数据源后,单击右侧删除或批量删除按钮 ,删除勾选中的数据源。
上下游任务查看:数据源登记解析出的上下游任务,您可对上游依赖任务进行编辑、添加、删除、禁用等操作;下游任务仅支持查看。
可填写 Flink 相关的动态参数和执行参数,具体参数设置详见 Flink 官方文档。
单行编辑模式:填写 key-value,key值只允许字母、数字、小数点、下划线和连字符。
a. 添加一行参数
b. 删除当前这行参数
c. 清空输入框中已输入的参数值
脚本编辑模式:通过 JSON、Yaml 的格式填写运行参数。
任务所需参数配置完成后,将任务提交发布到运维中心实时任务运维中执行。
单击操作栏中的保存和提交上线按钮,在弹窗中,需先通过提交事前检查和提交上线等上线流程,最后单击确认按钮,完成作业提交。详见概述---流式任务提交发布。
注意
上线流程中的“提交事前检查”,需租户主账号或项目管理员先在项目控制台 > 流水线管理中,创建相应的流水线检查事宜后方可显示。详见流水线管理。
后续任务运维操作详见:实时任务运维。