Serverless Flink SQL

1 概述

DataLeap接入了流式计算 Flink 版，在关联 Flink 的项目和资源池后，可以进行 Flink 作业开发。可以通过 Serverless Flink SQL 作业实现不同存储系统之间的 ETL 等。本文以一个简单的示例，将为您介绍 Serverless Flink SQL作业相关的开发流程操作。

2 使用前提

1. DataLeap产品需开通 DataOps敏捷研发、大数据分析、数据开发特惠版或分布式数据自治服务后，才可绑定流式计算 Flink 引擎。绑定引擎操作详见：项目管理。

2. 子账号操作项目绑定 Flink 引擎实例时：

   1. 主账号需要先在流式计算 Flink 版控制台导入 IAM 用户。操作详见：用户管理。

   2. 并将子账号加入到对应引擎项目中。操作详见：引擎项目成员管理。

3. 目前 Flink 版本只支持 1.11。

3 任务配置说明

3.1 新建任务

1. 登录DataLeap租户控制台。

2. 在概览界面，显示加入的项目中，单击数据开发进入对应项目。

3. 在任务开发界面，左侧导航栏中，单击新建任务按钮，进入新建任务页面。

4. 选择任务类型：

   1. 分类：数据开发。

   2. 绑定引擎：流式计算 Flink 版。

   3. 关联引擎项目：默认选择引擎绑定时选择的引擎项目，不可更改。

   4. 选择任务：流式数据 Serverless Flink SQL。

5. 填写任务基本信息：

   1. 任务名称：输入任务的名称，只能由数字、字母、下划线、-和.组成, 首尾只能是数字、字母，且允许输入 1~63 个字符。

   2. 保存至： 选择任务存放的目标文件夹目录。

6. 单击确认按钮，成功创建任务。

3.2 编辑任务

新建任务成功后，进入代码开发编辑界面，通过 DDL 和 DML 编辑 SQL ，示例代码如下：
详细语法可参考Flink 1.11 官方文档。

3.2.1 编辑SQL代码

以下为示例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;

3.2.2 数据源格式

SQL 编辑完成后，您可在下方选择数据源格式：

• 支持选择其他和 Pb 数据类型。

• Pb 类定义：需要将 Pb 类定义文件拖动到输入框中，或手动输入，如以下示例：

  1. 一次只能有一个入口类。

  2. 若入口类下的字段有嵌套其他类，这些类必须放在一起定义。

  填写示例：
  syntax = "proto2";
  package abase_test;
  message AbaseTest {
  required int64 first_id = 1;
  required int64 latest_id = 2;
  }
  

• Pb 入口 message：填写类名信息

3.3 导航栏功能区解析

3.4 参数设置

单击右侧导航栏中参数设置，进行任务的基本信息、任务输入参数、资源设置、数据源登记、Flink 运行参数配置。

3.4.1 基本信息

Serverless Flink SQL 任务的基本信息配置如下：

3.4.2 任务输入参数

您可通过手动添加的方式，选择项目参数或自定义参数，以变量或常量形式传入流式任务中使用，来实现同一套代码在不同执行环境下能够自动替换参数来执行。添加操作详见 3.5.2-任务输入参数。

3.4.3 资源设置

设置任务运行时相关资源分配情况：

3.4.4 数据源登记

登记该任务使用的 Source 和 Sink 信息，以用于后续监控配置和血缘构建。支持通过自动解析和手动添加方式。

• 自动解析：保存 SQL 逻辑后，单击自动解析按钮，可根据 SQL 逻辑自动进行数据源上下游登记。

  • 源/目标：支持源/目标 Source 和 Sink 的选择。

  • 数据源类型：支持选择 DataGen、Rmq、Bmq、JDBC、Abase、火山/自建 Kafka 这些 Connector 的数据源自动解析。

    说明

    Abase 和 Rmq 数据源类型，仅支持解析操作，暂不支持上线发布后运行。且 Abase 数据源通常是在火山引擎内部业务上云中使用。

  • 各数据源类型需补充填写相应的信息，您可根据实际场景进行配置，如 Kafka 的数据源类型，需填写具体的Bootstrap Servers、Topic 名称、消费组等信息。

  • 直接上游任务：根据数据源匹配逻辑，自动解析出上游任务。您也可通过下方手动添加的方式，进行搜索上游任务名称，单击添加按钮，进行手动添加。

  • 直接下游任务：根据数据源匹配逻辑，自动解析出下游任务，但无法通过手动添加的方式进行操作，仅展现通过数据源匹配后的结果。

  • 自动解析的数据源，您可通过关闭自动解析按钮，将登记的数据源给删除。

  

• 手动添加：您也可以通过手动添加的方式进行数据源登记。

• 批量删除：手动添加的数据源，您可通过勾选已登记的数据源后，单击右侧删除或批量删除按钮 ，删除勾选中的数据源。

• 上下游任务查看：数据源登记解析出的上下游任务，您可对上游依赖任务进行编辑、添加、删除、禁用等操作；下游任务仅支持查看。

3.4.5 Flink 运行参数

可填写 Flink 相关的动态参数和执行参数，具体参数设置详见 Flink 官方文档。

• 单行编辑模式：填写 key-value，key值只允许字母、数字、小数点、下划线和连字符。
  a. 添加一行参数

  b. 删除当前这行参数

  c. 清空输入框中已输入的参数值
  

• 脚本编辑模式：通过 JSON、Yaml 的格式填写运行参数。

4 提交发布

任务所需参数配置完成后，将任务提交发布到运维中心实时任务运维中执行。
单击操作栏中的保存和提交上线按钮，在弹窗中，需先通过提交事前检查和提交上线等上线流程，最后单击确认按钮，完成作业提交。详见概述---流式任务提交发布。

后续任务运维操作详见：实时任务运维。