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

优采云发布时间: 2022-02-04 14:04

　　本文介绍如何使用腾讯云大数据组件完成实时监控系统的设计与实现。通过电话、微信消息等方式实时反馈监控报警信息，有效保障系统稳定运行。使用基于云的Kafka、Flink、ES等组件，大大降低了开发和运维人员的投入。

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

　　(二）程序架构

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

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

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

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

　　（七）安装和配置FilebeatFilebeat是一个轻量级的日志数据采集工具，通过监控指定位置的文件来采集信息。在这个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实现个性化监控。

　　(一）文件节拍采集数据

　　1、进入Filebeat根目录，启动Filebeat for data采集。例子中采集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数据。在 ES 控制台的 Kibana 页面查询数据，或者进入同一子网的 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}'

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

　　除了Stream Computing 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 自定义监视器

　　( )。