本文为您介绍Flink全托管SQL作业开发的限制说明和操作步骤。

使用限制

  • SQL编辑器提交的SQL作业,仅支持开源Flink V1.11和Flink V1.12版本。
  • SQL支持的上下游存储(Connector)列表,请参见支持的上下游存储

作业开发

为了方便您编写和管理Flink SQL作业,提高作业开发效率,Flink全托管产品为您提供Flink SQL的全套功能,包括元数据管理、UDF注册和SQL编辑器等。

  1. 登录Flink全托管开发控制台,新建作业。
    1. 登录实时计算控制台
    2. Flink全托管页签,单击目标工作空间操作列下的开发控制台
    3. 在左侧导航栏,单击作业开发
    4. 单击新建
    5. 新建文件对话框,填写作业配置信息。
      作业类型请选择为SQL。
      作业参数 说明
      文件名称 作业的名称。
      说明 作业名称在当前项目中必须保持唯一。
      作业类型 支持以下类型:
      • SQL
      • JAR
      • PYTHON
      部署目标 选择作业需要部署的集群,支持Per-Job集群(默认)和Session集群两种模式。两种集群模式有以下区别:
      • Per-Job集群(默认):作业之间资源隔离,每个作业都需要一个独立的JM,因此小任务JM的资源利用率较低。因此,适用于占用资源比较大或持续稳定运行的作业。
      • Session集群:多个作业可以复用相同的JM,可以提高JM资源利用率。因此适用于占用资源比较小或任务启停比较频繁的作业。
      说明 如果您需要开启SQL Preview功能,必须选择Session集群,且已将其设置为SQL Previews集群,详情请参见作业调试配置Session集群
    6. 单击确认
  2. 在作业开发页面,编写DDL和DML代码。
    示例代码如下。
    --创建源表datagen_source。
    CREATE TEMPORARY TABLE datagen_source(
      name VARCHAR
    ) WITH (
      'connector' = 'datagen'
    );
    --创建结果表blackhole_sink。
    CREATE TEMPORARY TABLE blackhole_sink(
      name  VARCHAR
    ) with (
      'connector' = 'blackhole'
    );
    
    --将源表数据插入到结果表。
    INSERT INTO blackhole_sink
    SELECT
      name
    from datagen_source;
  3. 在作业开发页面右侧,单击高级配置,填写配置信息。
    参数解释如下表所示。
    类别 配置项 说明
    基础配置 部署目标 您可以修改创建作业时已选择的部署目标。
    并行度 作业并发数。
    附加依赖文件 如果您需要添加更多依赖文件,请选择或者输入任意合法的文件地址。
    常规配置 Flink版本 支持Flink 1.10Flink 1.11Flink 1.12版本。
    Flink镜像标签 选择Flink镜像标签。
    使用native k8s方式提交 开启native k8s方式提交功能。
    Kerberos身份验证 填写文件地址或者上传文件,对ZooKeeper进行身份验证。
    行为配置 恢复策略 当作业状态变为运行时,State的恢复策略。取值如下:
    • Latest Savepoint :将从最新Savepoint文件恢复。
    • Latest State:将从最新的Savepoint或Checkpoint中恢复。
    • None:不带State恢复作业,即作业恢复后没有State了,需要重新开始计算了。
    说明 升级策略恢复策略需要配合使用。如果升级策略选择Stateful恢复策略选择None,则完成Savepoint后,作业会从作业的起始时间重新启动,而不是从Savepoint恢复。
    创建实例的最大重试次数 创建实例失败后的重试次数。
    Stop with Drain 如果开启Stop With Drain功能,当作业被手动停止或者保留State升级作业时,窗口中已有数据结果会输出,即使没有满足关窗条件。
    Flink配置 Checkpoint间隔 定时执行Checkpoint的时间间隔。如果不填写,将会关闭 Checkpoint。
    两次Checkpoint之间的最短时间间隔 两次Checkpoint之间的最短时间间隔,如果Checkpoint最大并行度是1,则该配置确保两个Checkpoint之间有一个最短时间间隔。
    Checkpoint Retention策略 当作业无法再重启或者作业被暂停时,是否需要保留最新完成的Checkpoint。该参数取值如下:
    • Always:作业停止时保留Checkpoint。
    • Only when FAILED:作业停止后删除Checkpoint。
    • Never(默认值):不保留Checkpoint。
    开启Unaligned Checkpoint 开启Unaligned Checkpoint会大大降低反压情况下Checkpoint的总执行时间。但是也会导致增大单次Checkpoint的大小。
    Flink重启策略配置 当有Task失败时,如果没有开启Checkpoint,JobManager进程不会重启。如果开启了Checkpoint,则JobManager进程会重启。该参数取值如下:
    • Failure Rate:基于失败率重启。
    • Fixed Delay:固定间隔重启。
    • No Restarts(默认值):不会重启。
    Flink Master 容错配置(高可用性) 该参数取值如下:
    • None(默认值):表示系统不会保存元数据给JobManager重启恢复时使用。
    • Kubernetes:表示系统会将元数据保存到Kubernetes 集群中供JobManager重启恢复使用。
    更多Flink配置 在此设置其他Flink配置。例如taskmanager.numberOfTaskSlots: 1
    资源配置 Task Managers数量 默认与并行度一致。
    Job Manager CPUs 默认值为1。
    Job Manager Memory 最小值为500 Mi。单位建议使用Gi或Mi,例如,1024 Mi或1.5 Gi。
    Task Manager CPUs 默认值为1。
    Task Manager Memory 最小值为1 Gi。单位建议使用Gi或Mi,例如,1024 Mi或1.5 Gi。
    日志配置 Root Log Level TRACE、DEBUG、INFO、WARN和ERROR。
    Log Levels 填写日志名称和日志级别。
    Logging Profile 日志模板,可以选择系统模板,也可以选择用户配置。
  4. 单击保存
  5. 单击验证
  6. 验证通过后,单击运行
  7. 单击上线
    完成作业开发和语法检查后,即可上线作业,即可将数据发布至生产环境。