搭建实时报警平台

在本实践用户可以了解到如何搭建一个简单的实时报警平台，该平台将应用多个云服务，结合数据湖工厂（DLF）的作业编辑和作业调度功能来实现。

假设客户有一个数据中心部署了很多应用，需要建立统一的运维系统，实时接收应用的告警信息。

• 当告警级别达到严重及以上级别时，向用户发送一条消息。

• 每天提供一个运维报表，统计各应用的告警级别数据。

为解决以上场景的需求，我们设计了如下方案：

图1 方案设计

操作流程如下：

1. 实时数据导入：通过数据接入服务（DIS）将数据中心的告警数据实时导入到实时流计算服务（CS）。

2. 数据清洗和预处理：CS对告警数据进行数据清洗和预处理。

3. 发送告警消息：当告警级别超过指定值时向用户发送短信。

4. 数据导出和存储：清洗过的数据进入DIS通道，DIS根据导入时间将告警数据按日期存放到OBS。

5. 输出告警统计报表：通过DLI SQL脚本建立外部分区数据表，以及按照告警分区时间和告警类别进行统计。

6. 迁移数据：告警统计表计算完成后，将数据通过云数据迁移服务（CDM）统一导出到RDS MySQL数据库。

环境准备

• 已开通对象存储服务（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语句实现的功能：

   关键参数说明：

   
   

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

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

1. 在OBS的“s3a://dlfexample/alarm_info”目录下，根据日期新建DLI分区。假设当前日期为2018/10/10，那么在“s3a://dlfexample/alarm_info”目录下新建“2018/10/09”的DLI分区，用于存放前一天的数据表。

2. 按照告警分区时间和告警类别进行统计，将统计结果插入alarm_count_info数据表。

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中构建一个作业每天自动执行，那么就可以每天输出告警统计报表、每天自动迁移数据。

1. 创建一个DLF批处理作业，作业名称为“job_alarm”。

   
   

   图13 创建DLF作业
   
   

   
   

2. 然后进入到作业开发页面，拖动DLI SQL和CDM Job节点到画布中，连接并配置节点的属性。

   
   

   图14 连接和配置节点属性
   
   

   关键说明：

   
   

• dli_partition_count（DLI SQL节点）：在节点属性中，关联使用DLI SQL脚本开发构建告警报表脚本中开发完成的DLI SQL脚本“dli_partition_count”。

• obs_rds（CDM Job节点）：在节点属性中，关联创建CDM作业中创建的CDM作业“obs_rds”。

3. 作业编排完成后，单击，测试运行作业。

4. 如果日志运行正常，单击画布空白处，在右侧的“调度配置”页面，配置作业的调度策略。

   
   

   图15 调度配置
   
   

   说明：

   
   

• 2018/10/10至2018/11/09，每天2点执行一次作业。

5. 最后我们需要保存作业（单击），并执行调度作业（单击），实现作业每天自动运行。