高效数据传输：轻松上手将Kafka实时数据接入CnosDB

本篇我们将主要介绍如何在 Ubuntu 22.04.2 LTS 环境下，实现一个Kafka+Telegraf+CnosDB 同步实时获取流数据并存储的方案。在本次操作中，CnosDB 版本是2.3.0，Kafka 版本是2.5.1，Telegraf 版本是1.27.1 

随着越来越多的应用程序架构转向微服务或无服务器结构，应用程序和服务的数量每天都在增加。用户既可以通过实时聚合，也可以通过输出为测量或指标的计算，来处理数量不断增加的时间序列数据。面对产生的海量数据，用户可以通过多种方式来捕获和观察系统中数据的变化，在云原生环境中，最流行的一种是使用事件。

Apache Kafka是一个耐用、高性能的消息系统，也被认为是分布式流处理平台。它可应用于许多用例，包括消息传递、数据集成、日志聚合和指标。而就指标而言，仅有消息主干或代理是不够的。虽然 Apache Kafka 很耐用，但它并不是为运行指标和监控查询而设计的。这恰恰正是 CnosDB 的长处。

架构方案

通过将这Kafka、Telegraf和CnosDB 三者结合起来，可以实现数据的完整流程：

  1. 数据生成：使用传感器、设备或其他数据源产生数据，并将其发送到Kafka主题。
  2. Kafka 消息队列：Kafka 接收并存储数据流，确保数据安全和可靠性。
  3. Telegraf 消费者：Telegraf 作为 Kafka 的消费者，订阅 Kafka 主题并获取数据流。
  4. CnosDB 数据存储：经过预处理的数据由 Telegraf 发送到 CnosDB 中进行时序数据的存储。

整体的应用程序架构如图所示：

图片

Kafka

Apache Kafka 是一个开源分布式流处理平台，它被设计用于处理实时数据流，具有高可靠性、高吞吐量和低延迟的特点，目前已经被大多数公司使用。它的使用方式非常多样化，包括：

  • 流处理：它通过存储实时事件以进行聚合、丰富和处理来提供事件主干。
  • 指标：Apache Kafka 成为许多分布式组件或应用程序（例如微服务）的集中聚合点。这些应用程序可以发送实时指标以供其他平台使用，包括 CnosDB。
  • 数据集成：可以捕获数据和事件更改并将其发送到 Apache Kafka，任何需要对这些更改采取行动的应用程序都可以使用它们。
  • 日志聚合：Apache Kafka 可以充当日志流平台的消息主干，将日志块转换为数据流。

几个核心概念

  1. 实例（Broker）：Kafka的Broker是Kafka集群中的服务器节点，负责存储和转发消息，提供高可用性、容错性和可靠性。
  2. 主题（Topic）：Apache Kafka 中的 topic ，是逻辑存储单元，就像关系数据库的表一样。主题通过分区通过代理进行分发，提供可扩展性和弹性。
  3. 生产者（Producer）：生产者将消息发布到Kafka的指定主题。生产者可以选择将消息发送到特定的分区，也可以让Kafka自动决定分配策略。
  4. 消费者（Consumer）：消费者从指定主题的一个或多个分区中读取消息。消费者可以以不同的方式进行组织，如单播、多播、消费者组等。
  5. 发布-订阅模式：是指生产者将消息发布到一个或多个主题，而消费者可以订阅一个或多个主题，从中接收并处理消息。

简单来说就是，当客户端将数据发送到 Apache Kafka 集群实例时，它必须将其发送到某个主题。

此外，当客户端从 Apache Kafka 集群读取数据时，它必须从主题中读取。向 Apache Kafka 发送数据的客户端成为生产者，而从 Kafka 集群读取数据的客户端则成为消费者。数据流向示意图如下：

图片

注：这里没有引入更复杂的概念，如topic分区、偏移量、消费者组等，用户可自行参考官方指导文档学习：

Kafka：【https://kafka.apache.org/documentation/#gettingStarted】

部署 Kafka

下载并安装Kafka【https://kafka.apache.org/】

1.前提：需确保有 JDK 环境和 Zookeeper 环境，如果没有可以使用下面的命令进行安装：

sudo apt install openjdk-8-jdk
sudo apt install zookeeper

2.下载 Kafka 安装包并解压

wget https://archive.apache.org/dist/kafka/2.5.1/kafka_2.12-2.5.1.tgz
tar -zxvf kafka_2.12-2.5.1.tgz

3.进入解压后的 Kafka 目录

cd  kafka_2.12-2.5.1

4.修改$KAFKA_HOME/config/server.properties的配置文件（可按需修改端口、日志路径等配置信息）

5.保存并关闭编辑器。运行下面的命令来启动Kafka：

bin/kafka-server-start.sh config/server.properties

Kafka 将在后台运行，并通过默认的 9092 端口监听连接。

Telegraf

Telegraf 是一个开源的服务器代理程序，用于收集、处理和传输系统和应用程序的指标数据。Telegraf 支持多种输入插件和输出插件，并且能够与各种不同类型的系统和服务进行集成。它可以从系统统计、日志文件、API 接口、消息队列等多个来源采集指标数据，并将其发送到各种目标，如 CnosDB 、Elasticsearch、Kafka、Prometheus 等。这使得 Telegraf 非常灵活，可适应不同的监控和数据处理场景。

  • 轻量级：Telegraf被设计为一个轻量级的代理程序，对系统资源的占用相对较小，可以高效运行在各种环境中。
  • 插件驱动：Telegraf使用插件来支持各种输入和输出功能。它提供了丰富的插件生态系统，涵盖了众多的系统和服务。用户可以根据自己的需求选择合适的插件来进行指标数据的采集和传输。
  • 数据处理和转换：Telegraf具有灵活的数据处理和转换功能，可以通过插件链（Plugin Chain）来对采集到的指标数据进行过滤、处理、转换和聚合，从而提供更加精确和高级的数据分析。

部署 Telegraf

1.安装 Telegraf

sudo apt-get update && sudo apt-get install telegraf

2.切换到 Telegraf 的默认配置文件所处目录 /etc/telegraf 下

3.在配置文件 telegraf.config 中添加目标 OUTPUT PLUGIN

[[outputs.http]]url = "http://127.0.0.1:8902/api/v1/write?db=telegraf"timeout = "5s"method = "POST"username = "root"password = ""data_format = "influx"use_batch_format = truecontent_encoding = "identity"idle_conn_timeout = 10

按需修改的参数：

url：CnosDB 地址和端口
username：连接 CnosDB 的用户名
password：连接 CnosDB 的用户名对应的密码

注：其余参数可与上述配置示例中保持一致

4.在配置文件中将下面的配置注释放开，可按需修改

[[inputs.kafka_consumer]]
brokers = ["127.0.0.1:9092"]
topics = ["oceanic"]
data_format = "json"

参数：

brokers：Kafka 的 broker list 
topics：指定写入 Kafka 目标的 topic
data_format：写入数据的格式

注：其余参数可与上述配置示例中保持一致

5.启动 Telegraf

telegraf -config /etc/telegraf/telegraf.conf

CnosDB

部署 CnosDB

详细操作请参考： CnosDB 安装

【https://docs.cnosdb.com/zh/latest/start/install.html】

整合

Kafka创建topic

1.进入 kafka 的 bin 文件夹下

2.执行命令，创建 topic

./kafka-topics.sh --create --topic oceanic --bootstrap-server localhost:9092 --partitions 1 --replication-factor 1

Python 模拟写入数据到Kakfa

1.编写代码：

import time
import json
import randomfrom kafka import KafkaProducerdef random_pressure():return round(random.uniform(0, 10), 1)def random_tempreture():return round(random.uniform(0, 100), 1)def random_visibility():return round(random.uniform(0, 100), 1)def get_json_data():data = {}data["pressure"] = random_pressure()data["temperature"] = random_temp_cels()data["visibility"] = random_visibility()return json.dumps(data) def main():producer = KafkaProducer(bootstrap_servers=['ip:9092'])for _ in rang(2000):json_data = get_json_data()producer.send('oceanic', bytes(f'{json_data}','UTF-8'))print(f"Sensor data is sent: {json_data}")time.sleep(5)if __name__ == "__main__":main()

2.运行Python脚本

python3 test.py

查看 kafka topic 中的数据

1.执行下面查看指定 topic 数据的命令

./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic oceanic --from-beginning

图片

查看同步到 CnosDB 中的数据

1.使用工具连接到CnosDB

cnosdb-cli

2.切换到指定库

\c public

3.查看数据

select * from kafka_consumer;

图片

补充阅读

1.使用 Telegraf 采集数据并写入 CnosDB:

https://docs.cnosdb.com/zh/latest/versatility/collect/telegraf.html

2.Python 连接器:

https://docs.cnosdb.com/zh/latest/reference/connector/python.html

3.CnosDB 快速开始:

https://docs.cnosdb.com/zh/latest/start/quick_start.html

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.mzph.cn/news/35078.shtml

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

无涯教程-Perl - redo函数

无涯教程-Perl - redo函数

描述 此函数将重新启动当前循环,而不会强制判断控制语句。块中不再执行任何语句。如果存在继续块,将不会执行。如果指定了LABEL,则在LABEL标识的循环开始时重新开始执行。 语法 以下是此函数的简单语法- redo LABELredo返回值 此函数不返回任何值。 例 以下是显示其基本…
阅读更多...
用友时空KSOA SQL注入漏洞复现（HW0day）

用友时空KSOA SQL注入漏洞复现（HW0day）

0x01 产品简介 用友时空KSOA是建立在SOA理念指导下研发的新一代产品，是根据流通企业最前沿的I需求推出的统一的IT基础架构，它可以让流通企业各个时期建立的IT系统之间彼此轻松对话，帮助流通企业保护原有的IT投资，简化IT管理&#…
阅读更多...
以商业大数据技术助力数据合规流通体系建立，合合信息参编《数据经纪从业人员评价规范》团标

以商业大数据技术助力数据合规流通体系建立，合合信息参编《数据经纪从业人员评价规范》团标

经国务院批准，由北京市人民政府、国家发展和改革委员会、工业和信息化部、商务部、国家互联网信息办公室、中国科学技术协会共同主办的2023 全球数字经济大会于近期隆重召开。由数交数据经纪（深圳）有限公司为主要发起单位，合合信息…
阅读更多...
深度剖析堆栈指针

深度剖析堆栈指针

为什么打印root的值与&root->value的值是一样的呢 测试结果： *号一个变量到底取出来的是什么？ 以前我写过一句话，就是说，如果看到一个*变量，那就是直逼这个变量所保存的内存地址，然后取出里面保存的…
阅读更多...
Skeleton-Aware Networks for Deep Motion Retargeting

Skeleton-Aware Networks for Deep Motion Retargeting

Skeleton-Aware Networks for Deep Motion Retargeting解析 摘要1. 简介2. Related Work2.1 运动重定向（Motion Retargeting）2.2 Neural Motion Processing 3. 概述（Overview）4. 骨骼感知深度运动处理4.1 运动表征4.2 骨架卷积4.3…
阅读更多...
Spring Boot + Vue3前后端分离实战wiki知识库系统＜十二＞--用户管理单点登录开发一

Spring Boot + Vue3前后端分离实战wiki知识库系统＜十二＞--用户管理单点登录开发一

目标： 在上一次Spring Boot Vue3前后端分离实战wiki知识库系统＜十一＞--文档管理功能开发三我们已经完成了文档管理的功能模块开发，接下来则开启新模块的学习---用户登录，这块还是有不少知识点值得学习的，…
阅读更多...
机器人CPP编程基础-02变量Variables

机器人CPP编程基础-02变量Variables

机器人CPP编程基础-01第一个程序Hello World 基础代码都可以借助人工智能工具进行学习。 C #include<iostream>using namespace std;main() {//Declaring an integer type variable A, allocates 4 bytes of memory.int A4;cout<<A <<endl;//Prints the a…
阅读更多...
Matlab绘制圆形（rectangle函数、viscircles函数和圆的参数方程）

Matlab绘制圆形（rectangle函数、viscircles函数和圆的参数方程）

基于matlab绘制圆形 一、rectangle函数 对于绘制圆心坐标为&#xff08;x&#xff0c;y&#xff09;半径为r的圆形&#xff0c;函数为&#xff1a; x0; y0; r1; rectangle(Position, [x-r,y-r,2*r,2*r], Curvature, [1 1],EdgeColor, r); axis equalEdgeColor表示颜色 二、…
阅读更多...
解决lldb调试时可能出现的personality set failed: Function not implemented

解决lldb调试时可能出现的personality set failed: Function not implemented

最近在尝试使用Visual Studio 2022远程连接Linux进行C/C的开发&#xff0c;由于CentOS风波不断&#xff0c;所以现在的开发基本上都是使用ubuntu了&#xff0c;但是目前VS2022有一些BUG&#xff0c;就是远程调试时&#xff0c;如果目标系统是ubuntu则会出现启动调试器很慢的问题…
阅读更多...
学习笔记整理-JS-03-表达式和运算符

学习笔记整理-JS-03-表达式和运算符

[[toc]] 一、表达式和运算符 1. 表达式 表达式种类 算术、关系、逻辑、赋值、综合 二、JS基本表达式 1. 算术运算符 意义运算符加减-乘*除/取余% 加减乘除 加减的符号和数学一致&#xff0c;乘号是*号&#xff0c;除法是/号默认情况&#xff0c;乘除法的优先级高于加法和…
阅读更多...
安卓源码分析（10）Lifecycle实现组件生命周期管理

安卓源码分析（10）Lifecycle实现组件生命周期管理

参考&#xff1a; https://developer.android.google.cn/topic/libraries/architecture/lifecycle?hlzh-cn#java https://developer.android.google.cn/reference/androidx/lifecycle/Lifecycle 文章目录 1、概述2、LifeCycle类3、LifecycleOwner类4、LifecycleObserver类 1、…
阅读更多...
nginx禁用3DES和DES弱加密算法

nginx禁用3DES和DES弱加密算法

nginx禁用3DES和DES弱加密算法 项目背景 最近护网行动&#xff0c;收到漏洞报告&#xff0c;如下&#xff1a; 漏洞名称SSL/TLS协议信息泄露漏洞(CVE-2016-2183)【原理扫描】详细描述TLS是安全传输层协议&#xff0c;用于在两个通信应用程序之间提供保密性和数据完整性。 TLS…
阅读更多...
opencv 基础50-图像轮廓学习03-Hu矩函数介绍及示例-cv2.HuMoments()

opencv 基础50-图像轮廓学习03-Hu矩函数介绍及示例-cv2.HuMoments()

什么是Hu 矩&#xff1f; Hu 矩&#xff08;Hu Moments&#xff09;是由计算机视觉领域的科学家Ming-Kuei Hu于1962年提出的一种图像特征描述方法。这些矩是用于描述图像形状和几何特征的不变特征&#xff0c;具有平移、旋转和尺度不变性&#xff0c;适用于图像识别、匹配和形状…
阅读更多...
Docker查看、创建、进入容器相关的命令

Docker查看、创建、进入容器相关的命令

1.查看、创建、进入容器的指令 用-it指令创建出来的容器&#xff0c;创建完成之后会立马进入容器。退出之后立马关闭容器。 docker run -it --namec1 centos:7 /bin/bash退出容器&#xff1a; exit查看现在正在运行的容器命令&#xff1a; docker ps查看历史容器&#xff0…
阅读更多...
docker小白第二天

docker小白第二天

centos上安装docker docker官网&#xff0c;docker官网&#xff0c;找到下图中的doc文档。 进入如下页面 选中manuals&#xff0c;安装docker引擎。 最终centos下的docker安装文档链接&#xff1a;安装文档链接. 具体安装步骤&#xff1a; 1、打开Centos&#xff0c;输入命…
阅读更多...
【BASH】回顾与知识点梳理（十五）

【BASH】回顾与知识点梳理（十五）

【BASH】回顾与知识点梳理 十五 十五. 指令与文件的搜寻15.1 脚本文件名的搜寻which (寻找『执行档』) 15.2 文件档名的搜寻whereis (由一些特定的目录中寻找文件文件名)locate / updatedbfind与时间有关的选项与使用者或组名有关的参数与文件权限及名称有关的参数额外可进行的…
阅读更多...
JVM垃圾回收

JVM垃圾回收

如何确定垃圾 对堆垃圾回收前的第一步就是要判断哪些对象已经死亡&#xff08;即不能再被任何途径使用的对象&#xff09; 引用计数法 这个方法就是为对象添加计数器来标识引用个数&#xff0c;计数器为 0 的对象就是不可能再被使用的。但是这种方法存在循环引用问题&#x…
阅读更多...
布谷鸟配音：一站式配音软件

布谷鸟配音：一站式配音软件

这是一款智能语音合成软件&#xff0c;可以快速将文字转换成语音&#xff0c;拥有多种真人模拟发音&#xff0c;可以选择不同男声、女声、童声&#xff0c;以及四川话、粤语等中文方言和外语配音&#xff0c;并且可对语速、语调、节奏、数字读法、多音字、背景音等进行全方位设…
阅读更多...
学习笔记整理-JS-01-语法与变量

学习笔记整理-JS-01-语法与变量

文章目录 一、语法与变量1. 初识JavaScript2. JavaScript的历史3. JavaScript与ECMAScript的关系4. JavaScript的体系5. JavaScript的语言风格和特性 二、语法1. JavaScript的书写位置2. 认识输出语句3. REPL环境&#xff0c;交互式解析器4. 变量是什么5. 重点内容 一、语法与变…
阅读更多...
二、编写第一个 Spring MVC 程序

二、编写第一个 Spring MVC 程序

文章目录 一、编写第一个 Spring MVC 程序 一、编写第一个 Spring MVC 程序 代码示例 创建 maven 项目&#xff0c;以此项目为父项目&#xff0c;在父项目的 pom.xml 中导入相关依赖 <dependencies><dependency><groupId>junit</groupId><artifactI…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023