大数据-Storm流式框架（二）--wordcount案例

一、编写wordcount案例

1、新建java项目

2、添加storm的jar包

storm软件包中lib目录下的所有jar包

3、编写java类

WordCountTopology.java
package com.bjsxt.storm.wc;import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.generated.StormTopology;
import backtype.storm.topology.TopologyBuilder;
import backtype.storm.tuple.Fields;
import backtype.storm.utils.Utils;public class WordCountTopology {public static void main(String[] args) {// 拓扑封装了计算逻辑TopologyBuilder builder = new TopologyBuilder();// 设置lineSpout：该spout负责向拓扑发送句子builder.setSpout("lineSpout", new LineSpout());// 设置切分闪电，该闪电处理从水龙头lineSpout通过随机分组发送过来的元组builder.setBolt("splitBolt", new SplitBolt()).shuffleGrouping("lineSpout");// 定义一个计数闪电，该闪电从splitBolt闪电通过按字段分组的方式分发过来的元组// 按照元组中word的值进行分组。要保证相同的单词一定发送给同一个闪电。builder.setBolt("countBolt", new CountBolt()).fieldsGrouping("splitBolt", new Fields("word"));// 通过建造者创建一个拓扑的实例StormTopology wordCountTopology = builder.createTopology();// 本地模拟集群LocalCluster cluster = new LocalCluster();Config config = new Config();// 将拓扑提交到本地模拟集群cluster.submitTopology("wordCountTopology", config, wordCountTopology);// 睡眠10s，也就是让本地模拟集群运行10sUtils.sleep(10000);// 关闭本地模拟集群cluster.shutdown();}}
LineSpout.java
package com.bjsxt.storm.wc;import backtype.storm.spout.SpoutOutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichSpout;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;
import backtype.storm.utils.Utils;import java.util.Map;public class LineSpout extends BaseRichSpout {private SpoutOutputCollector collector;private String[] lines = {"The logic for a realtime application is packaged into a Storm topology","A stream is an unbounded sequence of tuples that is processed and created in parallel in a distributed fashion","A spout is a source of streams in a topology","Bolts can do anything from filtering, functions, aggregations, joins, talking to databases, and more.","A stream grouping defines how that stream should be partitioned among the bolt's tasks.","Storm guarantees that every spout tuple will be fully processed by the topology","Each spout or bolt executes as many tasks across the cluster","Each worker process is a physical JVM and executes a subset of all the tasks for the topology"};private int index = 0;@Overridepublic void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {// 在该组件在集群中初始化的时候调用一次this.collector = collector;}@Overridepublic void nextTuple() {// 由storm的线程不停地调用，以便从数据源获取元组// 该方法不需要自己写循环和遍历// 该方法不能阻塞// 负责从数据源获取元组，向DAG发送元组// 轮询取出句子String lingStr = lines[index % lines.length];// 将句子封装为元组发射collector.emit(new Values(lingStr));index++;Utils.sleep(10);}@Overridepublic void declareOutputFields(OutputFieldsDeclarer declarer) {// 用于声明元组的结构以及流
//        declarer.declareStream("s1", new Fields("key1", "key2", "key3"));
//        declarer.declareStream("s2", new Fields("key21", "key22"));// 发送元组的时候就有一个字段，是line，它的值是句子// 可以将元组想象为map集合，只不过其key是固定的几个declarer.declare(new Fields("line"));}
}
SplitBolt.java
package com.bjsxt.storm.wc;import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;import java.util.Map;public class SplitBolt extends BaseRichBolt {private OutputCollector collector;@Overridepublic void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {this.collector = collector;}@Overridepublic void execute(Tuple input) {String lineStr = input.getStringByField("line");String[] wordStrs = lineStr.split(" ");for (String wordStr : wordStrs) {// <hello, 1>this.collector.emit(new Values(wordStr, 1));}}@Overridepublic void declareOutputFields(OutputFieldsDeclarer declarer) {declarer.declare(new Fields("word", "count"));}
}
CountBolt.java
package com.bjsxt.storm.wc;import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Tuple;import java.util.HashMap;
import java.util.Map;public class CountBolt extends BaseRichBolt {private Map<String, Integer> counts;@Overridepublic void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {counts = new HashMap<>();}@Overridepublic void execute(Tuple input) {
//        new Fields("word", "count")String wordStr = input.getStringByField("word");Integer count = input.getIntegerByField("count");Integer sum = counts.get(wordStr);if (sum == null) {counts.put(wordStr, count);} else {counts.put(wordStr, sum + count);}counts.forEach((k, v) -> {System.out.println(k + "_________" + v);});System.out.println("========================================");}@Overridepublic void declareOutputFields(OutputFieldsDeclarer declarer) {}
}

4、运行

右键运行WordCountTopology

二、Storm整体架构

Storm配置项

说明

java.library.path

Storm本身依赖包的路径，存在多个时用冒号分隔

storm.local.dir

Storm使用的本地文件系统目录（必须存在并且storm进程可读写）。默认是storm的根目录下的storm-local。

storm.zookeeper.servers

storm集群对应的zookeeper集群的主机列表

storm.zookeeper.port

storm集群对应的zookeeper集群的服务端口，zookeeper默认端口为2181

storm.zookeeper.root

storm的元数据在zookeeper中存储的根目录，默认值是/storm

storm.cluster.mode

storm运行模式，local或distributed。集群模式需设置为distributed

storm.messaging.transport

storm的消息传输机制，使用netty作为消息传输时设置为backtype.storm.messaging.netty.Context

nimbus.host

整个storm集群的nimbus节点

nimbus.supervisor.timeout.secs

storm中每个被发射出去的消息处理的超时时间，该时间影响到消息的处理，同时在storm ui上杀掉一个拓扑时的默认时间（kill动作发出后多长时间才会真正将该拓扑杀掉）。默认值是60

ui.port

storm自带UI，以http服务形式支持访问，此处设置该http服务的端口（非root用户端口号需要大于1024）

ui.childopts

storm UI进程的java参数设置（对java进程的约束都可以在此设置，如内存等）

logviewer.port

此处用于设置该Log Viewer进程的端口（Log Viewer进程也是http形式，需要运行在每个storm节点上）。默认值8000

logviewer.childopts

Log Viewer进程的参数设置

logviewer.appender.name

storm log4j的appender，设置的名字对应于文件storm/log4j2/cluster.xml中设置的appender，cluster.xml可以控制storm logger的级别

supervisor.slots.ports

storm的slot，最好设置为OS核数的整数倍；同时由于storm是基于内存的实时计算，slot数不要大于每台物理机可运行slot个数：（物理内存-虚拟内存）/单个java进程最大可占用内存数

worker.childopts

storm的worker进程的java限制，有效地设置该参数能够在拓扑异常时进行原因分析：

-Xms1024m -Xmx1024m -XX:+UseConcMarkSweepGC -XX:+UseCMSInitiatingOccupancyOnly -XX:CMSInitiatingOccupancyFraction=70 -XX:+HeapDumpOnOutOfMemoryError

其中：Xms为单个java进程最小占用内存数，Xmx为最大占用内存数，设置HeapDumpOnOutOfMemoryError的好处是，当内存使用量超过Xmx时，java进程将被JVM杀掉同时会生成java_pid<pid数字>.hprof文件，使用MemoryAnalyzer分析hprof文件将能分析出内存使用情况从而进行相应的调整、分析是否有内存溢出等情况

storm.messaging.netty.buffer_size

netty传输的buffer大小，默认为5MB，当spout发射的消息较大时，此处需要对应调整

storm.messaging.netty.max_retries

这几个参数是关于使用netty作为底层消息传输时的相关设置，需要重视，否则可能由于bug而引起错误:

java.lang.IllegalArgumentException: timeout value is negative

storm.messaging.netty.max_wait_ms

storm.messaging.netty.min_wait_ms

topology.debug

该参数可以在拓扑中覆盖，表示该拓扑是否运行于debug模式。运行于debug模式时，storm将记录拓扑中收发消息等的详细信息，线上环境不建议打开

topology.acker.executors

storm通过acker机制保证消息不丢失，此参数用于设置每个拓扑的acker数量，由于acker基本消耗的资源较小，强烈建议将此参数设置在较低的水平，可以在拓扑中进行覆盖

topology.max.spout.pending

一个spout任务中处于pending状态的最大元组数量。该配置应用于单个任务，而不是整个spout或拓扑，可在拓扑中进行覆盖。

此外，storm/log4j2/cluster.xml文件中可以配置storm的日志级别矩阵信息等。

操作系统的配置，其中有两项需要配置（通过ulimit -a查看）：

1、open files：当前用户可以打开的文件描述符数。

2、max user processes：当前用户可以运行的进程数，此参数太小将引起storm的一个错误：

java.lang.OutOfMemoryError: unable to create new native thread

部署注意事项：

  1. 在storm根目录下有一个lib目录，存放storm本身依赖的jar包，此处的所有jar会被storm worker进行启动时加载，个人编写的jar包不能放在此处，以免包更新带来不便
  2. 向storm集群提交拓扑时，建议将该拓扑所有依赖的jar包和业务源代码打到一个jar包中（fat jar），如此则业务需要的jar包都和拓扑在同一个jar包中，否则当拓扑依赖的jar包更新时需要将该更新包放到所有的storm节点上。如果是在一个集群中，fat jar可以保证不同业务的jar包是独立的，不会混淆。

nimbus

  1. 接收jar包：提交应用拓扑
  2. 任务分配：将拓扑的任务分配给worker
  3. 资源调度：监控各个supervisor节点的状态进行负载均衡等。
  4. Nimbus不需要像supervisor节点那么高的配置，storm ui也不需要高配置，可以和nimbus节点运行在同一台服务器节点上。

supervisor

  1. 监听nimbus的任务分配，启动分配到的worker来对相应的任务进行处理。
  2. 监控本地的worker进程，如果发现状态不正常会杀死worker并重启，超过一定次数后将分配给该错误状态的worker的任务交还给nimbus进行再次分配。
  3. 删除本地不再运行的任务

worker

完成拓扑中定义的业务逻辑，即执行拓扑的进程。

一个worker的基本执行步骤：

  1. 根据zookeeper中拓扑的组件分配变化，创建或移除worker到worker的链接
  2. 创建executor（执行器）的输入队列receive-queue-map和输出队列transfer-queue
  3. 创建worker的接收线程receive-thread和发送线程transfer-thread
  4. 根据组件分配关系创建executor
    1. executor即worker JVM进程中的一个java线程，一般默认每个executor负责执行一个task任务
  5. 在executor中执行具体的任务（spout或者bolt）来执行具体的业务逻辑。
    1. 检查需要运行的task信息
    2. 获取相应的task信息，即spout/bolt信息

每个任务对应一个线程或多个任务对应一个线程

线程称为executor

executor在worker中运行

worker是一个JVM进程

在supervisor中运行

worker中的数据流：

worker中线程间通信使用的是Disruptor，进程间通信可能是netty也可以是zmq。默认使用netty。

数据流：

  1. 每个worker绑定一个socket端口作为数据的输入，此端口作为socket的服务器端一直监听运行。
  2. 根据拓扑的关系，确定需要向外通信的任务所在的worker地址，并同该worker也创建好socket连接，此时该worker是作为socket的客户端。
  3. receive thread负责将每个executor所需要的数据放入对应的receive-queue-map中，然后由executor来获取自己所需要的数据，这个过程通过disruptor进行通信。
  4. executor执行完操作需要对外发送数据时，首先kryo将数据序列化，然后通过disruptor将数据放入对外的transfer-queue中。
  5. transfer thread完成数据的发送工作。
  6. 如果executor需要对外发送的数据接收方和executor在同一个worker节点，则不需要执行序列化操作，调用disruptor的publish方法直接放到接收方的executor对应的队列中即可。

与MapReduce架构的对比：

提交作业过程

  1. 客户端提交拓扑代码到nimbus的nimbus/inbox目录下。
  2. nimbus对topology进行校验、处理
  3. nimbus针对该拓扑建立本地目录：nimbus/stormdist/topology-id

该目录下有三个文件：

    1. stormjar.jar 从nimbus/inbox移动来的topology的jar包
    2. stormcode.ser 对topology对象的序列化
    3. stormconf.ser topology的运行配置信息
  2. nimbus的调度器根据拓扑的配置计算task，并把task分配到不同的worker上，调度的结果写入zookeeper的/task节点下。
  3. zookeeper上建立assignments节点，存储task和supervisor中worker的对应关系。
  4. zookeeper上创建workerbeats节点监控worker的心跳。
  5. supervisor去zookeeper上获取分配的task信息，启动一个或多个worker来执行。
  6. 每个worker上运行多个task，task由executor来执行。
  7. worker根据拓扑信息初始化建立task之间的连接
  8. 相同worker内的task通过DisruptorQueue通信，不同worker间默认采用netty通信

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

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

相关文章

Pytorch代码入门学习之分类任务（三）：定义损失函数与优化器

Pytorch代码入门学习之分类任务（三）：定义损失函数与优化器

一、定义损失函数 1.1 代码 criterion nn.CrossEntropyLoss() 1.2 损失函数简介 神经网络的学习通过某个指标表示目前的状态&#xff0c;然后以这个指标为基准&#xff0c;寻找最优的权重参数。神经网络以某个指标为线索寻找最优权重参数&#xff0c;该指标称为损失函数&am…
阅读更多...
IP地址规划的基本方法

IP地址规划的基本方法

IP地址规划是构建和管理网络的关键步骤&#xff0c;它涉及到为网络中的设备分配合适的IP地址&#xff0c;以确保网络的高效性、安全性和可管理性。在本文中&#xff0c;我们将探讨IP地址规划的基本方法&#xff0c;以帮助网络管理员和工程师更好地设计和维护网络。 1. 理解IP地…
阅读更多...
基于JAVA的天猫商场系统设计与实现，springboot+jsp，MySQL数据库，前台用户+后台管理，完美运行，有一万五千字论文

基于JAVA的天猫商场系统设计与实现，springboot+jsp，MySQL数据库，前台用户+后台管理，完美运行，有一万五千字论文

目录 演示视频 基本介绍 论文目录 系统截图 演示视频 基本介绍 基于JAVA的天猫商场系统设计与实现&#xff0c;springbootjsp&#xff0c;MySQL数据库&#xff0c;前台用户后台管理&#xff0c;完美运行&#xff0c;有一万五千字论文。 本系统在HTML和CSS的基础上&#xf…
阅读更多...
Antv G6入门之旅--combo图

Antv G6入门之旅--combo图

目录 什么是AntV G6 G6 的特性 G6 文档 安装 1 在项目中使用 NPM 包引入 2 在 HTML 中使用 CDN 引入 使用 Step 1 创建容器 Step 2 数据准备 Step 3 创建关系图 Step 4 配置数据源&#xff0c;渲染 React 中使用 G6 Combo图 什么是AntV G6 G6 是一个图可视化引擎…
阅读更多...
OpenCV学习（二）——OpenCV中绘图功能

OpenCV学习（二）——OpenCV中绘图功能

2. OpenCV中绘图功能2.1 画线2.2 画矩形2.3 画圆2.4 画多边形2.5 添加文本 2. OpenCV中绘图功能 绘图可以实现画线、画矩形、画圆、画多边形和添加文本等操作。 import cv2 import numpy as np# 读取图像 img cv2.imread(lena.jpg)# 画直线 cv2.line(img, (0, 0), (512, 512…
阅读更多...
还不知道光场相机吗？

还不知道光场相机吗？

1.什么是光场&#xff1f; 光场&#xff08;light field&#xff09;&#xff1a;就是指光在每一个方向通过每一个点的光量。 从概念里&#xff0c;你至少可以得到两点信息&#xff1a; 光场包含光的方向光场包含一个点的光量 2.什么是光场相机 我们知道普通的相机拍照成像…
阅读更多...
Windows环境下Apache安装部署说明及常见问题解决

Windows环境下Apache安装部署说明及常见问题解决

一、软件准备 1.1 Python的下载与安装 见博客 链接: Python下载安装 1.2 Pycharm的下载与安装 见博客 链接: pycharm安装 1.3 Mysql的下载与安装 见博客 链接: MySQL安装 1.4 Navicat的下载与安装 可参考软件安装管家。 解释说明:Pycharm是Python的集成编译环境&#xff0c;Nav…
阅读更多...
SpringBoot2.7.14整合redis7

SpringBoot2.7.14整合redis7

需要的依赖库&#xff1a; <dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><optional>true</optional></dependency><dependency><groupId>org.springframework.boot</gro…
阅读更多...
中文编程开发语言工具编程实际案例：台球棋牌混合计时计费软件使用的编程构件说明

中文编程开发语言工具编程实际案例：台球棋牌混合计时计费软件使用的编程构件说明

中文编程开发语言工具编程实际案例&#xff1a;台球棋牌混合计时计费软件使用的编程构件说明 上图说明&#xff1a;该软件可以用于桌球和棋牌同时计时计费&#xff0c;在没有开台的时候&#xff0c;图片是处于等待状态&#xff0c;这使用编程工具中的固定图像构件&#xff0c;在…
阅读更多...
基于Java的音乐网站管理系统设计与实现(源码+lw+部署文档+讲解等)

基于Java的音乐网站管理系统设计与实现(源码+lw+部署文档+讲解等)

文章目录 前言具体实现截图论文参考详细视频演示为什么选择我自己的网站自己的小程序&#xff08;小蔡coding&#xff09; 代码参考数据库参考源码获取 前言 &#x1f497;博主介绍&#xff1a;✌全网粉丝10W,CSDN特邀作者、博客专家、CSDN新星计划导师、全栈领域优质创作者&am…
阅读更多...
vue3中使用svg并封装成组件

vue3中使用svg并封装成组件

打包svg地图 安装插件 yarn add vite-plugin-svg-icons -D # or npm i vite-plugin-svg-icons -D # or pnpm install vite-plugin-svg-icons -D使用插件 vite.config.ts import { VantResolver } from unplugin-vue-components/resolvers import { createSvgIconsPlugin } from…
阅读更多...
操作系统【OS】I/O核心子系统

操作系统【OS】I/O核心子系统

定义 举例 用户层软件 用户层软件实现了与用户交互的接口用户层软件将用户请求翻译成格式化的I/O请求&#xff0c;并通过“系统调用”请求操作系统内核的服务用户可直接使用该层提供的、与I/0操作相关的库函数对设备进行操作 如发送read命令如讲二进制整数转换为ascii码的…
阅读更多...
基于opencv的selenium滑动验证码的实现

基于opencv的selenium滑动验证码的实现

这篇文章主要介绍了基于opencv的selenium滑动验证码的实现&#xff0c;文中通过示例代码介绍的非常详细&#xff0c;对大家的学习或者工作具有一定的参考学习价值&#xff0c;需要的朋友们下面随着小编来一起学习学习吧 基于selenium进行动作链 由于最近很多人聊到滑动验证码…
阅读更多...
解决javascript报错：SyntaxError: Invalid Unicode escape sequence

解决javascript报错：SyntaxError: Invalid Unicode escape sequence

在处理cookie时报这个错&#xff1a; 网上搜了一圈都说是反斜杠问题&#xff0c;要把\替换成/ 但是试了网上的replace替换&#xff1a; replace(/\\/g, /) 结果没有用&#xff01;&#xff01;&#xff01; 然后我干脆直接做了一个最简单字符串赋值&#xff0c;再打印出来。…
阅读更多...
C++ stack 的使用

C++ stack 的使用

目录 1. 无参构造函数 2. void push(const T& x) 3. void pop() 4. T& top() 5. bool empty() 6. size_t size() 7. 总结 1. stack是一种容器适配器&#xff0c;专门用在具有后进先出操作的上下文环境中&#xff0c;其删除只能从容器的一端进行 元素的插入与…
阅读更多...
【机器学习合集】人脸表情分类任务Pytorch实现TensorBoardX的使用 -＞（个人学习记录笔记）

【机器学习合集】人脸表情分类任务Pytorch实现TensorBoardX的使用 -＞（个人学习记录笔记）

人脸表情分类任务 注意&#xff1a;整个项目来自阿里云天池&#xff0c;下面是开发人员的联系方式&#xff0c;本人仅作为学习记录&#xff01;&#xff01;&#xff01;该文章原因&#xff0c;学习该项目&#xff0c;完善注释内容&#xff0c;针对新版本的Pytorch进行部分代码…
阅读更多...
快速入门Elasticsearch：安装、基本概念、分词器和文档基本操作详解

快速入门Elasticsearch：安装、基本概念、分词器和文档基本操作详解

本文主要介绍快速入门 Elasticsearch&#xff0c;从 安装 、 基本概念 、 分词器 、*** 文档基本操作 *** 这 4 个方面快速入门。 Elasticsearch 是一款近实时的搜索引擎&#xff0c;底层是基于 Lucene 做搜索&#xff0c;再此基础上加入了分布式的特性&#xff0c;以便支持海…
阅读更多...
2022年12月 Python（一级）真题解析#中国电子学会#全国青少年软件编程等级考试

2022年12月 Python（一级）真题解析#中国电子学会#全国青少年软件编程等级考试

Python等级考试（1~6级）全部真题・点这里 一、单选题（共25题，每题2分，共50分） 第1题 关于Python语言的注释，以下选项中描述错误的是？（ ） A: Python语言有两种注释方式：单行注释和多行注释 B: Python语言的单行注释以#开头 C: Python多行注释使用###来做为标记 D: …
阅读更多...
RabbitMQ高级篇 笔记

RabbitMQ高级篇 笔记

这是一些高级的内容。 RabbitMQ还是运行在网络上的&#xff0c;倘若遇到了网络故障&#xff0c;mq自己挂了&#xff0c;出异常了&#xff0c;都会造成最终状态不一致的问题。这就是可靠性问题。 可靠性&#xff1a;一个消息发送出去之后&#xff0c;至少被消费1次。 要解决这3个…
阅读更多...
十九、类型信息（1）

十九、类型信息（1）

本章概要 为什么需要 RTTI RTTI&#xff08;RunTime Type Information&#xff0c;运行时类型信息&#xff09;能够在程序运行时发现和使用类型信息 RTTI 把我们从只能在编译期进行面向类型操作的禁锢中解脱了出来&#xff0c;并且让我们可以使用某些非常强大的程序。对 RTTI …
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023