You need to enable JavaScript to run this app.
导航
EMR Flink SQL
最近更新时间:2023.12.11 10:57:26首次发布时间:2021.08.13 15:07:14

1 概述

流式 EMR SQL 底层的计算引擎为 Apache Flink,其符合标准 SQL 语义,降低了用户进行实时开发的门槛,支持在线创建、解析和运维流式任务。
本文将为您介绍 EMR Flink SQL 任务的相关使用。

2 使用前提

  1. DataLeap产品需开通数据开发特惠版、DataOps敏捷研发、大数据分析分布式数据自治服务后,才可创建火山引擎 E-MapReduce(EMR)流式数据开发任务。

  2. EMR 引擎绑定的集群类型、版本及依赖的服务,需满足以下条件之一,方可创建 EMR Flink SQL 任务:

    支持集群版本支持集群类型依赖集群服务
    EMR-1.3.1HadoopFlink
    FlinkFlink
    EMR-3.2.1及以上HadoopFlink 和 GTS
  3. EMR Flink SQL 目前仅支持原生Connector:kafka、datagen、print。

  4. DataLeap 项目控制台首次绑定 EMR 集群时,会提示在 EMR 集群关联的安全组中添加 8898 和 9030 端口,您单击确定按钮即可实现自动添加。添加后,为确保能在 DataLeap 上正常进行数据开发和执行任务,需保证相关端口一直存在于安全组中,不要删除。

    详见创建项目

3 源数据准备

  1. 新建Kafka Topic,具体操作请参考:创建Topic

  2. 获取 bootstrap.servers,具体操作请参考:使用默认接入点连接实例

4.1 新建任务

选择流式数据 > EMR Flink SQL 任务类型新建任务。

  1. 登录DataLeap租户控制台

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

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

  4. 选择任务类型:

    1. 分类:数据开发。

    2. 绑定引擎:EMR

    3. 关联实例:在下拉列表中选择项目绑定时的 Hadoop 引擎实例,项目支持绑定多个 EMR 引擎,您可在多个引擎实例中选取该任务需执行的引擎实例。多引擎绑定详见创建项目

    4. 选择任务:流式数据 EMR Flink SQL

  5. 填写任务基本信息:

    1. 任务名称:输入任务的名称,只允许字符.、字母、数字、下划线、连字符、[]、【】、()、()以及中文字符,且需要在127个字符以内。

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

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

4.2 编辑任务

新建任务成功后,进入代码开发编辑界面,通过 DDL 和 DML 编写 SQL,下图以DataGen生成数据写入Kafka为例进行说明,详细语法可参考对应版本 Flink 官方文档

CREATE  TABLE datagen_source (f_sequence INT)
        WITH ('connector' = 'datagen', 'rows-per-second' = '60');

CREATE  TABLE kafka_sink (f_sequence INT, f_ts BIGINT)
        WITH (
            'connector' = 'kafka',
            'topic' = '{{kafka_topic}}',
            'properties.bootstrap.servers' = '{{bootstrap.servers}}',
            'format' = 'csv',
            'scan.startup.mode' = 'earliest-offset'
        );

INSERT INTO kafka_sink
SELECT  f_sequence,
        UNIX_TIMESTAMP() AS f_ts
FROM    datagen_source;

4.3 导航栏功能区解析

注意

任务被模板使用后,格式化、表管理、执行引擎选择功能将不支持使用。

功能名称描述
格式化依据在个性化设置中的 SQL 格式化风格的设置,格式化书写的代码,使其语法结构看起来简洁明了。
解析解析检查书写的 SQL 代码的语法和语义正确性,运行前检查语法错误信息,防止运行出错。

任务模板

您可以选择是否通过任务模板方式,便捷快速的复用代码模板逻辑,在弹窗中选择 EMR Flink SQL 任务模板,并选择相应的版本号,输入替换的参数即可完成复用。

注意

  • 使用任务模板时,界面编辑器不支持再次编辑,若当前任务模板代码,不符合您当前业务场景逻辑,您可通过修改或新建任务模板,形成新版本的任务模板,来实现不同业务场景的复用。
  • 修改模板时,需注意其他任务引用情况,操作说明详见:任务模板
执行引擎下拉选择 EMR 集群中 Flink 组件对应的执行引擎版本。

4.4 参数设置

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

4.4.1 基本信息

EMR Flink SQL 任务的基本信息配置如下:

参数名称描述
任务名称显示创建任务时输入的任务名称,参数设置中不支持修改,可以在左侧任务目录结构中的任务名称右侧更多单击重命名进行修改。
任务类型EMR Flink SQL
任务描述非必填,可对任务进行详细描述,方便后续查看和管理。

责任人

仅限一个成员,默认为任务创建人(任务执行失败、复查通过或者失败时的默认接收者),可根据实际需要,修改为其他项目成员。

  • 责任人需为项目中有编辑权限的成员,项目权限须由项目管理员授权。

  • DataLeap 通过项目空间实现开发协同,项目中具有编辑权限的角色,可对项目下的所有任务进行编辑,因此无须通过为任务设置多个责任人,来达到协同合作的效果。

Yarn 队列

支持下拉选择任务运行所需的 Yarn 队列信息。您可在创建项目 > 服务绑定 > EMR Yarn 队列绑定时,可添加绑定多个队列信息。详见创建项目

说明

EMR Yarn 队列的更多操作详见 YARN 队列管理

优先级

您可对流式任务设置任务优先级,指定当前任务的优先级情况:

  • 等级数字越小,代表优先级等级越高。

  • 其中 D3~D5 等级,您可直接在调度设置页面中设置。

标签

您可以自定义标签,用于标识某一类任务,以便快速搜索过滤,操作即时生效,无需重新上线任务。

  • 下拉选择项目归属的标签组,及对应的标签信息,支持添加多个标签组。

  • 若没有可选的标签组,您可进行以下操作步骤:
    a. 进入 DataLeap 智能市场 > 任务标签管理,进入任务标签管理控制台,新建标签组。详见任务标签管理

    b. 新建标签组成功后,前往项目管理 > 具体项目的配置信息 > 标签管理来添加标签组。详见标签管理

4.4.2 引入资源

Flink SQL 任务支持通过引入资源 Jar 包,并在 Jar 包中自定义 connector 的方式,进行 Flink SQL 任务的编辑、提交上线执行等操作。

说明

Jar 包中自定义 connector 和内置 connector 重名时,不支持在同一个任务中使用,否则任务执行时会出现冲突异常的情形。

在下拉框中,选择已通过资源库上传的 Jar 包资源,可支持多选。若您还未上传资源,您可单击“新建资源”按钮,前往资源库进行资源创建,操作详见资源库

4.4.3 资源设置

设置 EMR Flink SQL 运行时相关资源分配:

参数名称描述
TaskManager个数设置 flink 作业中 TaskManager 的数量
单TaskManagerCPU数设置单个 TaskManager 所占用的 CPU 数量
单TaskManager内存大小(MB)设置单个 TaskManager 所占用的内存大小
单TaskManager slot数设置单个 TaskManager 中 slot 的数量
JobManager CPU数设置单个 JobManager 所占用的 CPU 数量
JobManager内存设置单个 JobManager 所占用的内存大小

4.4.4 任务输入参数

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

4.4.5 数据源登记

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

  • 手动添加:单击手动添加按钮,可根据 SQL 逻辑进行数据源上下游登记。

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

    • 数据源类型:支持选择 DataGen、Rmq、Bmq、JDBC、Abase、火山/自建 Kafka 这些数据源类型。

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

    • 直接上游任务:您可通过手动输入任务名称方式,进行搜索上游任务,单击添加按钮,进行手动添加。

    • 直接下游任务:无法通过手动添加的方式进行操作,其余任务将该任务设置为上游时,此处会自动加载出下游任务。

    说明

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

    2. EMR 引擎暂不支持自动解析功能来登记数据源。

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

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

4.4.5 Flink 运行参数

可填写 Flink 相关的动态参数和执行参数,平台已为您提供一些常用的 SQL 参数、State 参数、Runtime 参数等,您可以根据实际情况进行选择,或者自行输入所需参数。更多参数详见 Flink 官方文档
参数配置操作详见 4.2 Flink 运行参数

5 提交任务

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

6 结果校验

流式任务启动后,可按照以下步骤,可以查看数据是否已写入目的端,以kafka为例,详细介绍请参考:消息查询

  1. 登录消息队列 Kafka版控制台

  2. 单击要查看的实例名称/ID,进入该实例详情页面。

  3. 单击上方导航栏中 Topic管理 页签,进入Kafka Topic管理页面。

  4. 单击要查询的 Topic名称,进入该实例页面,确认 末端Offset 是否有更新(即数值不为0)。

  5. 返回上一级后,单击消息查询页签,进入消息查询页面。

  6. 输入查询信息,下方列表中显示对应的消息。

  7. 单击消息所在行的下载消息,可下载该消息的信息。

  8. 将消息值进行Base64解码,确认结果和SQL逻辑是否匹配。