场景

一个数据中心部署了很多应用，需要建立统一的运维系统，实时接收应用的告警信息，如果告警级别严重以上，给用户发送一条信息，同时每天出一个运维报表，统计各应用各个级别告警的数据。

实现方案

数据中心的告警数据通过DIS实时导入CS实时流处理，CS对告警数据进行数据清洗和预处理，当告警级别超过一定级别时实时给用户发送短信。同时清洗过的数据进入DIS通道，DIS根据导入时间将告警数据按日期目录存放到OBS。

执行DLI SQL按天建DLI告警表分区，分区添加后统计告警记录到告警统计表。DLI SQL执行完成后，通过CDM统一导出到MYSQL数据库。以上操作过程在DLF服务上完成。DLF脚本开发中完成DLI SQL的开发。开发完成以后，DLI SQL节点关联开发的DLI SQL脚本。DLI SQL节点后连接CDM节点将数据导出到MYSQL数据库。最后配置DLF作业的调度周期为天，调度时间为凌晨2点。

实施步骤

1.创建2个DIS通道

①创建2个OBS目录，一个告警表目录alarm_info，一个告警统计表目录alarm_count_info。
②建立2个DIS通道，dis-alarm-input和dis-alarm-output。

③dis-alarm-output通道配置转储

2.创建SMN

①创建一个SMN主题alarm_over

②添加订阅，在订阅终端输入要接收通知的手机号码

3.创建CS作业

创建一个FLink SQL作业，在SQL编辑器中输入如下：

CREATE SOURCE STREAM alarm_info (
  alarm_id STRING,
  alarm_type INT
)
WITH (
  type = "dis",
  region = "cn-north-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-north-1",
  topic_urn = "urn:smn:cn-north-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-north-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;

提交购买，运行CS作业。
到这一步，使用DIS上传数据的工具上传数据，当告警超过级别可以收到短信，上传的全量告警数据在OBS上也看到了。OBS上的转储目录为/alarm_info/2018/08/31/。

4.使用DLF创建周期调度任务

①新建DLI连接
点击数据管理，在数据连接管理页面创建数据连接，选择数据连接类型为DLI。PS：已提前建好DLI集群。

②新建脚本
点击数据开发，新建脚本。输入脚本名称，脚本类型选择SQL，数据连接类型选择DLI。

在左侧选择DLI数据连接，资源队列，数据库。在SQL编辑框输入alarm_info的建表SQL语句。

执行脚本，刷新表列表，可以在表列表中看到新建的alarm_info表。
同样，新建脚本，执行alarm_count_info表建表语句。

alarm_info建表SQL
create table  alarm_info(alarm_time string, alarm_id string, alarm_type int ) using csv options(path 's3a://obs-fubei/alarm_info')
partitioned by(alarm_time)
alarm_count_info建表SQL
create table  alarm_info_count(alarm_time string, alarm_type int, alarm_count int)
using csv options(path 's3a://obs-fubei/alarm_info_count')

编写定时增加alarm_info表分区的SQL。表分区的键值和obs路径目录根据加载日期来获取。DLF脚本支持参数设置，在编写脚本的右侧点击参数设置，新增一个参数。输入$联想出目前DLF支持的系统函数。

选择getCurrentTime函数，getCurrentTime函数联想出参数：

$getCurrentTime(@@yyyyMMddHHmmss@@,-246060)。第一个参数是输出的格式，第二个参数是当前时间的偏移量，-246060为当前时间的前一天，60*60为当前时间加1小时。

一个小技巧，保存好的参数，可以直接在SQL编辑框，直接select 参数名，查看执行结果是否为用户想要的值。比如设置aa=$getCurrentTime(@@yyyyMMdd@@,-246060)

根据用户场景，DIS实时导入告警数据到OBS按天分目录/alarm_info/2018/08/31/。

新建一个周期调度脚本，每天凌晨2点在DIS导出到OBS目录上新建DLI分区。分区新增后，按照告警分区时间和告警类别进行统计，将统计结果插入alarm_info_count表。

新建两个参数，一个为alarm_info表分区值，一个为OBS分区目录路径。脚本参数和脚本内容如下：

dayParam：$getCurrentTime(@@yyyyMMdd@@,-24*60*60)
obsPathYear：$getCurrentTime(@@yyyy/MM/dd@@,-24*60*60)
ALTER TABLE alarm_info  ADD PARTITION (alarm_time = ${dayParam})
LOCATION 's3a://obs-fubei/alarm_info/${obsPathYear}';
insert into alarm_info_count
select alarm_time,alarm_type,count(alarm_type) from alarm_info where alarm_time = ${dayParam} group by alarm_time,alarm_type;

③新建CDM作业
在CDM集群中新建CDM作业从将OBS数据迁移到MYSQL。

④新建DLF作业
拖拉一个DLI SQL节点，选择要执行的DLI SQL脚本。再拖拉一个CDM节点，选择要执行的CDM作业。

将DLI SQL和CDM节点连接起来，执行完DLI SQL执行CDM作业。

配置作业的调度周期，每天凌晨2点执行一次。

保存启动调度。

⑤调度启动后，在运维调度的作业监控中可以看到作业正在运行中。

运行成功的作业可以在运行历史记录中看到。

运行失败的作业可以重跑那天的批次作业，查看日志分析失败原因。

一个简易的实时告警分析平台就构建完成了。

本文转载自华为云产品与解决方案公众号。

原文链接：https://mp.weixin.qq.com/s/Qw_UGDc7P2FVSvJPkWHZQQ

创作场景

使用数据湖工厂（DLF）搭建数据中心实时告警平台