文章实时采集(如何使用腾讯云大数据组件来完成实时监控系统的设计和实现)

优采云发布时间: 2021-10-23 11:11

　　本文介绍如何使用腾讯云大数据组件完成实时监控系统的设计与实现。通过实时采集和分析云服务器（CVM）及其App应用的CPU和内存消耗数据，使用短信、电话、微信等方式实时反馈监控报警信息，有效保证系统的稳定运行。使用基于云的Kafka、Flink、ES等组件，大大减少了开发运维人员的投入。

　　一、方案说明（一）概述本方案结合腾讯云CKafka、流计算Oceanus（Flink）、Elasticsearch、Prometheus等，通过Filebeat实时采集系统和应用监控数据，并传输到CKafka，将CKafka数据连接到Oceanus（Flink）进行流计算，通过简单的业务逻辑处理输出到Elasticsearch，最后通过Kibana页面查询结果。计划中使用Promethus监控系统指标，如作为流计算 Oceanus 作业运行状态，Cloud Grafana 监控 CVM 或业务应用指标。

　　(二）方案架构

　　二、前期准备在实施本方案之前，请确保已经创建并配置了相应的大数据组件。（一）创建私网VPC 私网VPC是您在腾讯云上自定义的逻辑隔离的网络空间，建议在搭建CKafka、流计算Oceanus、Elasticsearch集群等时选择同一个VPC服务.详情创建步骤请参考帮助文档()。（二）创建CKafka实例Kafka建议选择最新的2.4.1版本，与Filebeat兼容性更好< @采集工具购买完成后，创建一个Kafka主题：topic-app-info（三）创建流计算Oceanus集群流计算Oceanus是大数据产品生态系统的实时分析工具。基于Apache Flink，一站式开发，无缝连接、亚秒级延迟、低成本、安全稳定的企业级实时大数据分析平台。流计算Oceanus旨在实现企业数据价值最大化，加速企业实时数字化建设。在流计算中在Oceanus控制台的【集群管理】->【新建集群】页面创建集群。具体步骤请参考帮助文档（）。（四）在Elasticsearch控制台创建Elasticsearch实例，点击左上角【新建】创建集群，具体步骤请参考帮助文档（）。（五）

　　为了显示自定义的系统指标，需要购买 Promethus 服务。只需要自定义业务指标的同学可以省略这一步。

　　进入云监控控制台，点击左侧的【Prometheus监控】，新建一个Prometheus实例。具体步骤请参考帮助文档（）。

　　(六）创建独立的Grafana资源独立Grafana在灰色发布中，需要在Grafana管理页面()单独购买，实现业务监控指标的展示。

　　（七）安装配置Filebeat Filebeat是一个轻量级的日志数据工具采集，通过监控指定位置的文件来采集信息，在这个VPC下提供给需要监控主机的云服务器信息及应用信息安装Filebeat 安装方法一：下载Filebeat 下载地址(); 方法二：使用【Elasticsearch管理页面】->【beats管理】中提供的Filebeat，本例采用方法一。下载到CVM然后配置Filebeat，在filebeat.yml文件中添加如下配置项：

　　# 监控日志文件配置- type: logenabled: truepaths: - /tmp/test.log #- c:programdataelasticsearchlogs*

　　# 监控数据输出项配置output.kafka:version: 2.0.0 # kafka 版本号hosts: ["xx.xx.xx.xx:xxxx"] # 请填写实际的 IP 地址+端口topic: 'topic-app-info' # 请填写实际的 topic

　　请根据实际业务需求配置相应的Filebeat.yml文件，参考Filebeat官方文档()。

　　注：示例中使用的是2.4.1的CKafka版本，此处配置版本：2.0.0。不兼容的版本可能会出现“ERROR [kafka] kafka/client.go:341 Kafka (topic=topic-app-info): drop invalid message”错误

　　三、项目实现

　　接下来通过一个案例来介绍如何通过流计算Oceanus实现个性化监控。

　　(一）Filebeat 采集数据

　　1、进入Filebeat根目录，启动Filebeat获取数据采集。在示例中，采集显示了 top 命令中显示的 CPU、内存等信息。还可以采集jar应用日志、JVM使用情况、*敏*感*词*端口等，详情请参考Filebeat官网

　　()。

　　# filebeat 启动<br />./filebeat -e -c filebeat.yml<br /><br /># 监控系统信息写入 test.log 文件<br />top -d 10 >>/tmp/test.log

　　2、进入CKafka页面，点击左侧【消息查询】，查询对应的主题消息，验证数据是否为采集。

　　Filebeat采集中Kafka的数据格式：

　　{ "@timestamp": "2021-08-30T10:22:52.888Z", "@metadata": { "beat": "filebeat", "type": "_doc", "version": "7.14.0" }, "input": { "type": "log" }, "host": { "ip": ["xx.xx.xx.xx", "xx::xx:xx:xx:xx"], "mac": ["xx:xx:xx:xx:xx:xx"], "hostname": "xx.xx.xx.xx", "architecture": "x86_64", "os": { "type": "linux", "platform": "centos", "version": "7(Core)", "family": "redhat", "name": "CentOSLinux", "kernel": "3.10.0-1062.9.1.el7.x86_64", "codename": "Core" }, "id": "0ea734564f9a4e2881b866b82d679dfc", "name": "xx.xx.xx.xx", "containerized": false }, "agent": { "name": "xx.xx.xx.xx", "type": "filebeat", "version": "7.14.0", "hostname": "xx.xx.xx.xx", "ephemeral_id": "6c0922a6-17af-4474-9e88-1fc3b1c3b1a9", "id": "6b23463c-0654-4f8b-83a9-84ec75721311" }, "ecs": { "version": "1.10.0" }, "log": { "offset": 2449931, "file": { "path": "/tmp/test.log" } }, "message": "(B[m16root0-20000S0.00.00:00.00kworker/1:0H(B[m[39;49m[K"}

　　(二）创建 Flink SQL 作业

　　使用流计算 Oceanus 处理 CKafka 访问的数据并存储在 Elasticsearch 中。1、定义Source，根据Filebeat中json消息的格式构造Flink Table Source。

　　CREATE TABLE DataInput ( `@timestamp` VARCHAR, `host` ROW, `log` ROW, `message` VARCHAR) WITH ( 'connector' = 'kafka', -- 可选 'kafka','kafka-0.11'. 注意选择对应的内置 Connector 'topic' = 'topic-app-info', -- 替换为您要消费的 Topic 'scan.startup.mode' = 'earliest-offset', 'properties.bootstrap.servers' = '10.0.0.29:9092', 'properties.group.id' = 'oceanus_group2', -- 必选参数, 一定要指定 Group ID 'format' = 'json', 'json.ignore-parse-errors' = 'true', -- 忽略 JSON 结构解析异常 'json.fail-on-missing-field' = 'false' -- 如果设置为 true, 则遇到缺失字段会报错设置为 false 则缺失字段设置为 null);

　　2、定义*敏*感*词*

　　CREATE TABLE es_output ( `id` VARCHAR, `ip` ARRAY, `path` VARCHAR, `num` INTEGER, `message` VARCHAR, `createTime` VARCHAR) WITH ( 'connector.type' = 'elasticsearch', 'connector.version' = '6', 'connector.hosts' = 'http://10.0.0.175:9200', 'connector.index' = 'oceanus_test2', 'connector.document-type' = '_doc', 'connector.username' = 'elastic', 'connector.password' = 'yourpassword', 'update-mode' = 'upsert', -- 可选无主键的 'append' 模式，或有主键的 'upsert' 模式 'connector.key-null-literal' = 'n/a', -- 主键为 null 时的替代字符串，默认是 'null' 'format.type' = 'json' -- 输出数据格式, 目前只支持 'json');

　　3、处理业务数据

　　INSERT INTO es_outputSELECT host.id as `id`, host.ip as `ip`, log.file.path as `path`, log.`offset` as `num`, message, `@timestamp` as `createTime`from DataInput;

　　4、配置作业参数【内置连接器】选择flink-connector-elasticsearch6和flink-connector-kafka 注意：根据实际版本选择5、查询ES数据，在Kibana上查询数据ES控制台页面，或进入同一子网的CVM下，使用以下命令查询：

　　# 查询索引 username:password 请替换为实际账号密码curl -XGET -u username:password http://xx.xx.xx.xx:xxxx/oceanus_test2/_search -H 'Content-Type: application/json' -d'{ "query": { "match_all": {}}, "size": 10}'

　　更多访问方式请参考访问ES集群()。（三）系统指标监控本章主要实现系统信息监控，对Flink作业的运行状态进行监控和告警。Prometheus是一个非常灵活的时序数据库，通常用于监控数据的存储、计算和告警。流计算Oceanus建议用户使用腾讯云监控提供的Prometheus服务，避免部署和运维成本，同时还支持腾讯云的通知模板，可以通过短信、电话、邮件等方式轻松到达不同的报警信息、企业微信机器人等 Receiver. 监控配置流计算 Oceanus 作业监控

　　除了流计算Oceanus控制台自带的监控信息外，还可以配置目前支持任务级细粒度监控、作业级监控、集群Flink作业列表监控。1、流计算Oceanus作业详情页面，点击【作业参数】，在【高级参数】中添加如下配置：

　　pipeline.max-parallelism: 2048metrics.reporters: promgatewaymetrics.reporter.promgateway.host: xx.xx.xx.xx # Prometheus实例地址metrics.reporter.promgateway.port: 9090 # Prometheus实例端口metrics.reporter.promgateway.needBasicAuth: truemetrics.reporter.promgateway.password: xxxxxxxxxxx # Prometheus实例密码metrics.reporter.promgateway.interval: 10 SECONDS

　　2、在任何流计算Oceanus作业中，点击【云监控】进入云Prometheus实例，点击链接进入Grafana（灰度级Grafana这里不能输入），并导入json文件。详情请参考访问 Prometheus 自定义监视器

　　()。