【Kafka】Kafka Stream简单使用

一、实时流式计算

1. 概念

一般流式计算会与批量计算相比较。在流式计算模型中，输入是持续的，可以认为在时间上是无界的，也就意味着，永远拿不到全量数据去做计算。同时，计算结果是持续输出的，也即计算结果在时间上也是无界的。流式计算一般对实时性要求较高，同时一般是先定义目标计算，然后数据到来之后将计算逻辑应用于数据。同时为了提高计算效率，往往尽可能采用增量计算代替全量计算
在这里插入图片描述
流式计算就相当于上图的右侧扶梯，是可以源源不断的产生数据，源源不断的接收数据，没有边界。

2. 应用场景

  • 日志分析: 网站的用户访问日志进行实时的分析，计算访问量，用户画像，留存率等等，实时的进行数据分析，帮助企业进行决策
  • 大屏看板统计: 可以实时的查看网站注册数量，订单数量，购买数量，金额等。
  • 公交实时数据: 可以随时更新公交车方位，计算多久到达站牌等
  • 实时文章分值计算

比如应用较广的 头条类文章的分值计算，通过用户的行为实时文章的分值，分值越高就越被推荐。

3. Kafka Stream

近些年来，开源流处理领域涌现出了很多优秀框架。光是在 Apache 基金会孵化的项目，关于流处理的大数据框架就有十几个之多，比如早期的 Apache Samza、Apache Storm，以及这些年火爆的 Spark 以及 Flink 等。

3.1 Kafka Streams的特点

  • Kafka Stream提供了一个非常简单而轻量的Library，它可以非常方便地嵌入任意Java应用中，也可以任意方式打包和部署
  • 除了Kafka外，无任何外部依赖
  • 充分利用Kafka分区机制实现水平扩展和顺序性保证
  • 通过可容错的state store实现高效的状态操作（如windowed join和aggregation）
  • 支持正好一次处理语义
  • 提供记录级的处理能力，从而实现毫秒级的低延迟
  • 支持基于事件时间的窗口操作，并且可处理晚到的数据（late arrival of records）
  • 同时提供底层的处理原语Processor（类似于Storm的spout和bolt），以及高层抽象的DSL（类似于Spark的map/group/reduce）

在这里插入图片描述

3.2 关键概念

一个最简单的Streaming的结构如下图所示：
在这里插入图片描述

从一个Topic中读取到数据，经过一些处理操作之后，写入到另一个Topic中，这就是一个最简单的Streaming流式计算。其中，Source Topic中的数据会源源不断的产生新数据。
那么，我们再在上面的结构之上扩展一下，假设定义了多个Source Topic及Destination Topic，那就构成如下图所示的较为复杂的拓扑结构：
在这里插入图片描述

  • 源处理器（Source Processor）：源处理器是一个没有任何上游处理器的特殊类型的流处理器。它从一个或多个kafka主题生成输入流。通过消费这些主题的消息并将它们转发到下游处理器。
  • Sink处理器：sink处理器是一个没有下游流处理器的特殊类型的流处理器。它接收上游流处理器的消息发送到一个指定的Kafka主题。
    在这里插入图片描述
    Kafka Streams被认为是开发实时应用程序的最简单方法。它是一个Kafka的客户端API库，编写简单的java就可以实现流式处理。

3.3 KStream

KStream:数据结构类似于map，如下图，key-value键值对

在这里插入图片描述

KStream数据流（data stream），是一段顺序的，可以无限长，不断更新的数据集。
数据流中比较常记录的是事件，这些事件可以是一次鼠标点击（click），一次交易，或是传感器记录的位置数据。

KStream负责抽象的，就是数据流。与Kafka自身topic中的数据一样，类似日志，每一次操作都是向其中插入（insert）新数据。

二、测试kafkaStream

先看下简单的kafkaStream的KStream测试

需求分析：求单词个数（word count）
在这里插入图片描述

1. pom.xml引入依赖：

       <!-- kafka --><dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactId><exclusions><exclusion><groupId>org.apache.kafka</groupId><artifactId>kafka-clients</artifactId></exclusion></exclusions></dependency><dependency><groupId>org.apache.kafka</groupId><artifactId>kafka-clients</artifactId></dependency><dependency><groupId>com.alibaba</groupId><artifactId>fastjson</artifactId></dependency><dependency><groupId>org.apache.kafka</groupId><artifactId>kafka-streams</artifactId><exclusions><exclusion><artifactId>connect-json</artifactId><groupId>org.apache.kafka</groupId></exclusion><exclusion><groupId>org.apache.kafka</groupId><artifactId>kafka-clients</artifactId></exclusion></exclusions></dependency>

2. 配置文件

server:port: 9991
spring:application:name: kafka-demokafka:bootstrap-servers: 192.168.200.130:9092producer:retries: 10key-serializer: org.apache.kafka.common.serialization.StringSerializervalue-serializer: org.apache.kafka.common.serialization.StringSerializercompression-type: lz4consumer:group-id: ${spring.application.name}-testkey-deserializer: org.apache.kafka.common.serialization.StringDeserializervalue-deserializer: org.apache.kafka.common.serialization.StringDeserializer

3. 编写生产者

ProducerQuickStart.java 

package com.kafka.sample;import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.*;import java.util.Properties;@Slf4j
public class ProducerQuickStart {public static void main(String[] args) {//1. kafka的配置信息Properties prop = new Properties();//kafka的链接信息prop.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.200.130:9092");//配置重试次数prop.put(ProducerConfig.RETRIES_CONFIG, 5);//数据压缩prop.put(ProducerConfig.COMPRESSION_TYPE_CONFIG,"lz4");//ack配置  消息确认机制   默认ack=1,即只要集群首领节点收到消息，生产者就会收到一个来自服务器的成功响应
//        prop.put(ProducerConfig.ACKS_CONFIG,"all");消息key的序列化器prop.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");//消息value的序列化器prop.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");//2. 生产者对象KafkaProducer<String, String> producer = new KafkaProducer<String, String>(prop);//封装发送的消息ProducerRecord<String, String> producerRecord = new ProducerRecord<String, String>("itcast-topic-input", "key_001", "hello kafka");//3. 发送消息for (int i = 0; i < 5; i++) {producer.send(producerRecord);}//4. 关闭消息通道  必须关闭，否则消息发不出去producer.close();}
}

4 编写kafkaStream流式处理

KafkaStreamQuickStart.java 

package com.kafka.sample;import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.TimeWindows;
import org.apache.kafka.streams.kstream.ValueMapper;import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;/*** 流式处理*/
public class KafkaStreamQuickStart {public static void main(String[] args) {//kafka的配置信心Properties prop = new Properties();prop.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.200.130:9092");prop.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());prop.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());prop.put(StreamsConfig.APPLICATION_ID_CONFIG,"streams-quickstart");//stream 构建器StreamsBuilder streamsBuilder = new StreamsBuilder();//流式计算streamProcessor(streamsBuilder);//创建kafkaStream对象KafkaStreams kafkaStreams = new KafkaStreams(streamsBuilder.build(),prop);//开启流式计算kafkaStreams.start();}/*** 流式计算* 消息的内容：hello kafka  hello itcast* @param streamsBuilder*/private static void streamProcessor(StreamsBuilder streamsBuilder) {//创建kstream对象，同时指定从那个topic中接收消息KStream<String, String> stream = streamsBuilder.stream("itcast-topic-input");/*** 处理消息的value*/stream.flatMapValues(new ValueMapper<String, Iterable<String>>() {@Overridepublic Iterable<String> apply(String value) {return Arrays.asList(value.split(" "));}})//按照value进行聚合处理.groupBy((key,value)->value)//时间窗口.windowedBy(TimeWindows.of(Duration.ofSeconds(10)))//统计单词的个数.count()//转换为kStream.toStream().map((key,value)->{System.out.println("key:"+key+",vlaue:"+value);return new KeyValue<>(key.key().toString(),value.toString());})//发送消息.to("itcast-topic-out");}
}

5. 编写消费者

ConsumerQuickStart.java 

package com.kafka.sample;import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;import java.time.Duration;
import java.util.Collections;
import java.util.Properties;public class ConsumerQuickStart {public static void main(String[] args) {//1. 添加kafka的配置信息Properties properties = new Properties();// 配置链接信息properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.200.130:9092");//配置消费者组properties.put(ConsumerConfig.GROUP_ID_CONFIG, "group-2");//配置消息的反序列化器properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");//2. 消费者对象KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(properties);//3. 订阅主题consumer.subscribe(Collections.singletonList("itcast-topic-out"));//当前线程一直监听消息while(true){//4. 消费者拉取消息： 每秒拉取一次ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));for (ConsumerRecord<String, String> record : records) {System.out.println(record.key());System.out.println(record.value());}}}
}

启动项目：

  1. 在远端（192.168.200.130:9092）启动docker中的kafka容器
  2. 启动消费者ConsumerQuickStart的main函数
  3. 启动kafkastream的mian函数
  4. 启动生产者ProducerQuickStart的main函数

5. 控制台打印结果：

在这里插入图片描述
在这里插入图片描述

整个过程：
生产者向kafka中发送了5条“hello kafka”消息，topic均为itcast-topic-input。kafkastream监听这个topic，每10秒进行一次流式处理，将“hello kakfa”字符串分割，并统计每个单词出现的次数。然后转为kstream，发送消息到kafka中的topic=itcast-topic-out”。消费者监听“itcast-topic-out”的topic，消费消息。

三、Springboot整合kafkaStream

1. 配置文件新增

application.yml

server:port: 9991
spring:application:name: kafka-demokafka:bootstrap-servers: 192.168.200.130:9092producer:retries: 10key-serializer: org.apache.kafka.common.serialization.StringSerializervalue-serializer: org.apache.kafka.common.serialization.StringSerializercompression-type: lz4consumer:group-id: ${spring.application.name}-testkey-deserializer: org.apache.kafka.common.serialization.StringDeserializervalue-deserializer: org.apache.kafka.common.serialization.StringDeserializer
# kafkaStream新增以下配置
kafka:hosts: 192.168.200.130:9092group: ${spring.application.name}

2. 在微服务中新增配置类

KafkaStreamConfig.java 

package com.kafka.config;import lombok.Getter;
import lombok.Setter;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.StreamsConfig;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafkaStreams;
import org.springframework.kafka.annotation.KafkaStreamsDefaultConfiguration;
import org.springframework.kafka.config.KafkaStreamsConfiguration;import java.util.HashMap;
import java.util.Map;/*** 通过重新注册KafkaStreamsConfiguration对象，设置自定配置参数*/@Setter
@Getter
@Configuration
@EnableKafkaStreams
@ConfigurationProperties(prefix="kafka")
public class KafkaStreamConfig {private static final int MAX_MESSAGE_SIZE = 16* 1024 * 1024;private String hosts;private String group;@Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME)public KafkaStreamsConfiguration defaultKafkaStreamsConfig() {Map<String, Object> props = new HashMap<>();props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, hosts);props.put(StreamsConfig.APPLICATION_ID_CONFIG, this.getGroup()+"_stream_aid");props.put(StreamsConfig.CLIENT_ID_CONFIG, this.getGroup()+"_stream_cid");props.put(StreamsConfig.RETRIES_CONFIG, 10);props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());return new KafkaStreamsConfiguration(props);}
}

3. 使用kafkaStream监听消息

KafkaStreamHelloListener.java

package com.kafka.stream;import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.TimeWindows;
import org.apache.kafka.streams.kstream.ValueMapper;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;import java.time.Duration;
import java.util.Arrays;@Configuration
@Slf4j
public class KafkaStreamHelloListener {@Beanpublic KStream<String,String> kStream(StreamsBuilder streamsBuilder){//创建kstream对象，同时指定从那个topic中接收消息KStream<String, String> stream = streamsBuilder.stream("itcast-topic-input");stream.flatMapValues(new ValueMapper<String, Iterable<String>>() {@Overridepublic Iterable<String> apply(String value) {return Arrays.asList(value.split(" "));}})//根据value进行聚合分组.groupBy((key,value)->value)//聚合计算时间间隔.windowedBy(TimeWindows.of(Duration.ofSeconds(10)))//求单词的个数.count().toStream()//处理后的结果转换为string字符串.map((key,value)->{System.out.println("key:"+key+",value:"+value);return new KeyValue<>(key.key().toString(),value.toString());})//发送消息.to("itcast-topic-out");return stream;}
}

测试：

启动springboot应用程序，运行之前的ProducerQuickStart来生产消息，约10秒后，看到kafkaStream消息的处理结果
在这里插入图片描述

说明kafkaStream接收到消息并将多条消息进行了统一处理。

参考（推荐阅读）：

  1. https://cloud.tencent.com/developer/article/2100664
  2. https://www.cnblogs.com/tree1123/p/11457851.html

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

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

相关文章

向函数传递参数（传地址）

向函数传递参数（传地址）

过往课程 向函数传递参数&#xff08;传值、传引用、传const引用&#xff09; 传地址 向函数传地址&#xff0c;是指将变量的地址传递给函数。 函数通过声明参数为地址变量来接收一个变量的地址。 示例如下&#xff1a; #include <iostream> using namespace std;v…
阅读更多...
Mybatis 日志(JDK Log)

Mybatis 日志(JDK Log)

上一篇我们介绍了Mybatis中的参数&#xff0c;本篇我们使用JDK Log打印一下Mybatis运行时的日志&#xff0c;看一下Mybatis执行的过程。 这里我选取上一篇的示例进行JDK Log的集成&#xff0c;这里如果您想对上一篇进行详细了解&#xff0c;可以参考&#xff1a; Mybatis参数…
阅读更多...
C语言实现顺序表

C语言实现顺序表

顺序表 1.线性表 线性表是n个具有相同特性的数据元素的有限序列。 线性表是一种在实际中广泛使用的数据结构&#xff0c;常见的线性表&#xff1a;顺序表、链表、栈、队列、字符串… 线性表在逻辑上是线性结构&#xff0c;也就说是连续的一条直线。但是在物理结构上并不一定是…
阅读更多...
WPF怎么实现文件拖放功能winform怎么实现拖拽功能

WPF怎么实现文件拖放功能winform怎么实现拖拽功能

WPF怎么实现文件拖放功能winform怎么实现文件拖拽功能&#xff0c;在管理员模式下wpf winform怎么实现文件的拖拽功能 WPF实现文件拖放功能&#xff0c;正常情况并没有什么问题&#xff0c;但是如果你的程序使用管理员身份启动&#xff0c;你就会发现文件拖放功能就会失效。同…
阅读更多...
jmeter+nmon+crontab简单的执行接口定时压测

jmeter+nmon+crontab简单的执行接口定时压测

一、概述 临时接到任务要对系统的接口进行压测&#xff0c;上面的要求就是&#xff1a;压测&#xff0c;并发2000 在不熟悉系统的情况下&#xff0c;按目前的需求&#xff0c;需要做的步骤&#xff1a; 需要有接口脚本需要能监控系统性能需要能定时执行脚本 二、观察 >针…
阅读更多...
恒运资本：市盈率怎么算？

恒运资本：市盈率怎么算？

市盈率&#xff08;P/E ratio&#xff09;是判别一家公司股票价格合理性的一个重要目标&#xff0c;也是投资者评估公司股票投资价值的重要参阅目标。市盈率越高&#xff0c;表明相对于公司的收益来说&#xff0c;该公司的股票定价越高。市盈率越低&#xff0c;则表明该股票被低…
阅读更多...
App与小程序工具总结

App与小程序工具总结

文章目录 前言Burpsuite抓包问题LPosedJustTrustMe 绕过 SSL Pining小程序的反编译APP脱壳&#xff0c;反射大师、frida反射大师Frida 总结 前言 在进行渗透工作的时候&#xff0c;遇到过的App、小程序也不少了&#xff0c;有简单的&#xff0c;也有加固的比较不错的&#xff…
阅读更多...
技术深入解析与教程：网络安全技术探秘

技术深入解析与教程：网络安全技术探秘

第一章&#xff1a;引言 在当今数字化时代&#xff0c;网络安全已经成为了重要议题。随着各种信息和业务在网络上的传输与存储&#xff0c;安全问题也日益突出。本文将带您深入探讨网络安全领域中的关键技术&#xff0c;涵盖渗透测试、漏洞挖掘以及恶意软件分析等方面&#xf…
阅读更多...
PCD点云文件外部框框坐标计算

PCD点云文件外部框框坐标计算

PCD点云文件直接提取的是点云的坐标&#xff0c;不是最外面的box的坐标&#xff0c;因此可以通过&#xff1a; max_b octree.get_max_bound() min_b octree.get_min_bound()分别得到最大最小的xyz坐标&#xff0c;之后进行计算 点的序号和位置对应如下&#xff1a; 所有的…
阅读更多...
【数据结构与算法 模版】高频题刷题模版

【数据结构与算法 模版】高频题刷题模版

废话不多说&#xff0c;喊一句号子鼓励自己&#xff1a;程序员永不失业&#xff0c;程序员走向架构&#xff01;本篇Blog的主题是【】&#xff0c;使用【】这个基本的数据结构来实现&#xff0c;这个高频题的站点是&#xff1a;CodeTop&#xff0c;筛选条件为&#xff1a;目标公…
阅读更多...
软考A计划-网络工程师-复习背熟-路由器与交换配置和网络安全

软考A计划-网络工程师-复习背熟-路由器与交换配置和网络安全

点击跳转专栏>Unity3D特效百例点击跳转专栏>案例项目实战源码点击跳转专栏>游戏脚本-辅助自动化点击跳转专栏>Android控件全解手册点击跳转专栏>Scratch编程案例点击跳转>软考全系列点击跳转>蓝桥系列 &#x1f449;关于作者 专注于Android/Unity和各种游…
阅读更多...
ARM DIY（五）摄像头调试

ARM DIY（五）摄像头调试

前言 今天&#xff0c;就着摄像头的调试&#xff0c;从嵌入式工程师的角度&#xff0c;介绍如何从无到有&#xff0c;一步一步地调出一款设备。 摄像头型号&#xff1a;OV2640 开发步骤 分为 2 个阶段 5 个步骤 阶段一&#xff1a; 设备树、驱动、硬件 阶段二&#xff1a; 应…
阅读更多...
【跟小嘉学 Rust 编程】二十、进阶扩展

【跟小嘉学 Rust 编程】二十、进阶扩展

系列文章目录 【跟小嘉学 Rust 编程】一、Rust 编程基础 【跟小嘉学 Rust 编程】二、Rust 包管理工具使用 【跟小嘉学 Rust 编程】三、Rust 的基本程序概念 【跟小嘉学 Rust 编程】四、理解 Rust 的所有权概念 【跟小嘉学 Rust 编程】五、使用结构体关联结构化数据 【跟小嘉学…
阅读更多...
拿来即用修改密码功能

拿来即用修改密码功能

<template><div><!-- 重置密码 --><el-dialogtitle"修改密码"v-model"state.resetPwdDialogVisible":showClose"state.firstLogin ! 1"width"550px"close"onCancel":close-on-click-modal"false&…
阅读更多...
uniapp结合Canvas+renderjs根据经纬度绘制轨迹（二）

uniapp结合Canvas+renderjs根据经纬度绘制轨迹（二）

uniapp结合Canvasrenderjs根据经纬度绘制轨迹 文章目录 uniapp结合Canvasrenderjs根据经纬度绘制轨迹效果图templaterenderjsjs数据结构 ​ 根据官方建议要想在 app-vue 流畅使用 Canvas 动画&#xff0c;需要使用 renderjs 技术&#xff0c;把操作canvas的js逻辑放到视图层运…
阅读更多...
Python爬虫框架之Selenium库入门：用Python实现网页自动化测试详解

Python爬虫框架之Selenium库入门：用Python实现网页自动化测试详解

概要 是否还在为网页测试而烦恼&#xff1f;是否还在为重复的点击、等待而劳累&#xff1f;试试强大的Selenium&#xff01;让你的网页自动化测试变得轻松有趣&#xff01; 一、Selenium库到底是什么&#xff1f; Selenium 是一个强大的自动化测试工具&#xff0c;它可以让你直…
阅读更多...
华为 连接OSPF和RIP网络---OSPF和RIP网络相互引入

华为 连接OSPF和RIP网络---OSPF和RIP网络相互引入

路由引入简介 不同路由协议之间不能直接共享各自的路由信息&#xff0c;需要依靠配置路由的引入来实现。 获得路由信息一般有3种途径&#xff1a;直连网段、静态配置和路由协议。可以将通过这3种途径获得的路由信息引入到路由协议中&#xff0c;例如&#xff0c;把直连网段引入…
阅读更多...
Matlab图像处理-图像缩放

Matlab图像处理-图像缩放

基本概念 图像缩放是指将给定的图像在x轴方向按比例缩放a倍&#xff0c;在y轴方向按比例缩放b倍&#xff0c;从而获得一幅新的图像。 如果ab&#xff0c;即在x轴方向和y轴方向缩放的比率相同&#xff0c;则称这样的比例缩放为图像的全比例缩放。 如果a≠b&#xff0c;图像比…
阅读更多...
Nacos安装

Nacos安装

一、下载Nacos1.4.1二、单机版本安装 2.1 将下载的nacos安装包传输到服务器2.2 解压文件2.3 进入bin目录下 单机版本启动2.4 关闭nacos2.5 访问Nacos地址 IP&#xff1a;8848/nacos三、集群版本的安装 3.1 复制nacos安装包&#xff0c;修改为nacos8849&#xff0c;nacos8850&am…
阅读更多...
C# | DBSCAN聚类算法实现 —— 对直角坐标系中临近点的点进行聚类

C# | DBSCAN聚类算法实现 —— 对直角坐标系中临近点的点进行聚类

C# | DBSCAN聚类算法实现 聚类算法是一种常见的数据分析技术&#xff0c;用于将相似的数据对象归类到同一组或簇中。其中&#xff0c;DBSCAN&#xff08;Density-Based Spatial Clustering of Applications with Noise&#xff09;是一种基于密度的聚类算法&#xff0c;能够有效…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023