【最佳实践】搭建实时报警平台

环境准备

• 已开通对象存储服务（OBS），并创建桶，例如“s3a://dlfexample/alarm_info”、“s3a://dlfexample/alarm_count_info”，分别用于存放原始告警表和告警统计报表。
• 已开通云数据迁移服务（CDM），并创建集群“cdm-alarm”，用于创建CDM作业。
• 已开通数据湖探索服务（DLI）。
• 已开通消息通知服务（SMN）。

数据准备

原始告警表为数据中心的实时数据，包含告警ID、告警级别。示例数据如表1所示。

创建DIS通道

我们需要创建两个DIS通道，分别用于实时数据导入到CS、实时数据导出到OBS。

1. 创建实时数据导入到CS的通道，通道名称为“dis-alarm-input”。
   图2 创建input通道
   

2. 创建实时数据导出到OBS的通道，通道名称为“dis-alarm-output”。
   图3 创建output通道
   

   为dis-alarm-output通道配置转储任务，将通道中的数据按照导出时间转储到OBS的“s3a://dlfexample/alarm_info”目录下。

   图4 output通道配置转储任务
   

创建SMN主题

我们需要创建一个SMN主题并添加订阅，将需要收到告警通知的用户添加到订阅终端中。

1. 创建一个SMN主题，主题名称为“alarm_over”。
   图5 创建SMN主题
   

2. 为1中的主题添加订阅，指定告警消息类型和需要接收告警通知的用户。
   图6 添加订阅
   

   关键参数说明：

   • 协议：选择“短信”，当告警级别达到指定值时向用户发送短信通知。
   • 订阅终端：填写需要接收告警通知的用户手机号码。

使用CS作业管理构建告警通知工程

DIS通道（创建DIS通道）和SMN主题（创建SMN主题）创建完成后，我们就可以在CS中构建告警通知工程。

1. 在CS中创建一个Flink SQL作业，作业名称为“test”。
   图7 创建Flink SQL作业
   

2. 编辑1中创建的Flink SQL作业，在SQL编辑器中输入语句。
   图8 编辑Flink SQL作业
   

   SQL语句实现的功能：

   1. DIS通过工具上传实时数据至CS，使用1中创建的“dis-alarm-input”通道。
   2. 判断告警级别，当告警级别达到指定值时向用户发送短信通知，接收到短信通知的用户为2中配置的。
   3. CS处理过的数据再通过DIS导出到OBS中，使用2中创建的“dis-alarm-output”通道。

      CREATE SOURCE STREAM alarm_info (
        alarm_id STRING, 
        alarm_type INT
      )
      WITH (
        type = "dis",
        region = "cn-south-1",
        channel = "dis-alarm-input",
        partition_count = "1",
        encode = "csv",
        field_delimiter = ","
      );
      CREATE SINK STREAM over_alarm (
        alarm_over STRING /* over speed message */
      )
      WITH (
        type = "smn",
        region = "cn-south-1",
        topic_urn = "urn:smn:cn-south-1:6f2bf33af5104f45ab85de31d7841f5a:alarm_over",
        message_subject = "alarm",
        message_column = "alarm_over"
      );
      INSERT INTO over_alarm
      SELECT "your alarm over (" || CAST(alarm_type as CHAR(20)) || ") ."
      FROM alarm_info
      WHERE alarm_type > 8;
      CREATE SINK STREAM alarm_info_output (
        alarm_id STRING, 
        alarm_type INT
      )WITH (
        type ="dis",
        region = "cn-south-1",
        channel = "dis-alarm-output",
        PARTITION_KEY = "alarm_type",
        encode = "csv",
        field_delimiter = ","
      );
      INSERT INTO alarm_info_output 
      SELECT * 
      FROM alarm_info
      WHERE alarm_type > 0;

3. Flink SQL作业开发完成后，单击保存作业。
4. 保存作业后，单击，进入“作业配置清单”页面，单击“确认”，将作业提交并启动。

使用DLI SQL脚本开发构建告警报表脚本

我们需要通过SQL脚本在DLI中新建OBS表来存放数据表，然后再构建一个SQL脚本来统计告警信息。

1. 创建一个DLF到DLI的连接，数据连接名称为“dli”。
   图9 创建数据连接
   

2. 在DLI中创建一个数据库，用于存放数据表，数据库名称为“dlitest”。
   图10 创建数据库
   

3. 创建一个DLI SQL脚本，通过SQL语句来创建数据表alarm_info，alarm_count_info。

   其中，alarm_info、alarm_count_info都为OBS表，数据存储在OBS中，分别用于存放原始告警表、告警统计报表。

   图11 创建数据表
   

   关键操作说明：

   • 图11中的脚本开发区为临时调试区，关闭脚本页签后，开发区的内容将丢失。如需保留该SQL脚本，请单击，将脚本保存至指定的目录中。

   关键参数说明：

   • 数据连接：1中创建的DLI数据连接。
   • 数据库：2中创建的数据库。
   • 资源队列：使用DLI提供的默认资源队列“default”。
   • SQL语句：如下所示。

     create table  alarm_info(alarm_time string, alarm_id string, alarm_type int ) using csv options(path 's3a://dlfexample/alarm_info') partitioned by(alarm_time);
     create table  alarm_count_info(alarm_time string, alarm_type int, alarm_count int) using csv options(path 's3a://dlfexample/alarm_count_info');

4. 单击运行脚本，创建alarm_info、alarm_count_info数据表。
5. 清空编辑器中4的SQL语句，重新输入SQL语句。

   ALTER TABLE alarm_info  ADD PARTITION (alarm_time = ${dayParam}) 
   LOCATION 's3a://dlfexample/alarm_info/${obsPathYear}';
   insert into alarm_count_info 
   select alarm_time,alarm_type,count(alarm_type) from alarm_info where alarm_time = ${dayParam} group by alarm_time,alarm_type;

   SQL语句实现的功能：

   1. 在OBS的“s3a://dlfexample/alarm_info”目录下，根据日期新建DLI分区。假设当前日期为2018/10/10，那么在“s3a://dlfexample/alarm_info”目录下新建“2018/10/09”的DLI分区，用于存放前一天的数据表。
   2. 按照告警分区时间和告警类别进行统计，将统计结果插入alarm_count_info数据表。

   关键参数说明：

   • ${dayParam}：dayParam是指alarm_info表分区值，在脚本编辑器下方输入具体的参数值“$getCurrentTime(@@yyyyMMdd@@,-24*60*60)”。
   • ${obsPathYear}：obsPathYear是指OBS分区目录路径，在脚本编辑器下方输入具体的参数值“$getCurrentTime(@@yyyy/MM/dd@@,-24*60*60)”。

6. 脚本调试无误后，我们需要保存该脚本，脚本名称为“dli_partition_count”。在后续的作业中设置为定期执行该脚本（使用DLF作业开发和作业调度每天定时输出告警统计报表），实现定期输出告警统计报表。

创建CDM作业

方案的最后一步需要将OBS中的告警统计报表迁移到RDS MySQL中，我们选择使用CDM来实现该功能。

图12 创建CDM作业

关键参数说明：

• 作业名称：obs_rds，在后续的作业中设置为定期执行该作业（使用DLF作业开发和作业调度每天定时输出告警统计报表），实现定期迁移数据。
• 源端：存储告警统计报表的OBS目录，源连接“obs_link”需要提前在CDM中创建好。
• 目的端：即将存储告警统计报表的RDS MySQL空间，目的连接“mysql_link”需要提前在CDM中创建好。

使用DLF作业开发和作业调度每天定时输出告警统计报表

告警统计报表的脚本（使用DLI SQL脚本开发构建告警报表脚本）和数据迁移的CDM作业（创建CDM作业）创建完成后，我们在DLF中构建一个作业每天自动执行，那么就可以每天输出告警统计报表、每天自动迁移数据。