大数据-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.hqwc.cn/news/150746.html

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

相关文章

【Jenkins 安装】

【Jenkins 安装】

一&#xff1a;安装文件夹准备 在/home/admin 界面下新建三个文件夹&#xff0c;用来安装tomcat、maven 1.打开&#xff0c;/home/admin目录 cd /home/admin 2.新建三个文件夹 mkdir tomcat mkdir maven 二&#xff1a;安装tomcat 1.打开tomcat目录进行tomcat的安装 访问:h…
阅读更多...
时间、空间复杂度的例题详解

时间、空间复杂度的例题详解

文章前言 上篇文章带大家认识了数据结构和算法的含义&#xff0c;以及理解了时间、空间复杂度&#xff0c;那么接下来来深入理解一下时间、空间复杂度。 时间复杂度实例 实例1 // 计算Func2的时间复杂度&#xff1f; void Func2(int N) {int count 0;for (int k 0; k <…
阅读更多...
3DCAT+东风日产：共建线上个性化订车实时云渲染方案

3DCAT+东风日产：共建线上个性化订车实时云渲染方案

近年来&#xff0c;随着5G网络和云计算技术的不断发展&#xff0c;交互式3D实时云看车正在成为一种新的看车方式。 与传统的到4S店实地考察不同&#xff0c;消费者可以足不出户&#xff0c;通过网络与终端设备即可实现全方位展示、自选汽车配色、模拟效果、快捷选车并进行个性…
阅读更多...
【软考系统架构设计师】2023年系统架构师冲刺模拟习题之《软件工程》

【软考系统架构设计师】2023年系统架构师冲刺模拟习题之《软件工程》

在软考中软件工程模块主要包含以下考点&#xff1a; 文章目录 软件过程模型&#x1f31f;&#x1f31f;&#x1f31f;&#x1f31f;逆向工程&#x1f31f;基于构件的软件工程&#x1f31f;&#x1f31f;软件开发与软件设计与维护净室软件工程软件模型软件需求 软件过程模型&am…
阅读更多...
计算机网络文章荟萃

计算机网络文章荟萃

脑残式网络编程入门(二)&#xff1a;我们在读写Socket时&#xff0c;究竟在读写什么&#xff1f;-网络编程/专项技术区 - 即时通讯开发者社区! 1.什么是 socket - 掘金2.socket 的实现原理 - 掘金本文讲述了 socket 在 linux 操作系统下的数据结构&#xff0c;以及阻塞 IO 利用…
阅读更多...
RK3568-pcie接口

RK3568-pcie接口

pcie接口与sata接口 pcie总线pcie总线pcie控制器sata控制器nvme设备sata设备nvme协议ahci协议m-key接口b-key接口RC模式和EP模式 RC和EP分别对应主模式和从模式，普通的PCI RC主模式可以用于连接PCI-E以太网芯片或PCI-E的硬盘等外设。 RC模式使用外设一般都有LINUX驱动程序，安…
阅读更多...
RDMA概览

RDMA概览

RDMA(Remote Direct Memory Access&#xff0c;远程直接内存访问)&#xff0c;指能够访问(读写)远程机器的内存。有多种支持RDMA的网络协议&#xff0c;包括&#xff1a;Infiniband、RoCE和iWAPP。具体的API定义包含在内核文件linux/include/rdma/ib_verbs.h reference: 【精选…
阅读更多...
Elasticsearch聚合----aggregations的简单使用

Elasticsearch聚合----aggregations的简单使用

文章目录 Getting started1、搜索 address 中包含 mill 的所有人的年龄分布以及平均年龄&#xff0c;但不显示这些人的详情2、size0不展示命中记录&#xff0c;只展示聚合结果3、按照年龄聚合&#xff0c;并且请求这些年龄段的这些人的平均薪资4、查出所有年龄分布&#xff0c;…
阅读更多...
关于前端如何下载后端接口返回content-type为application/octet-stream的文件

关于前端如何下载后端接口返回content-type为application/octet-stream的文件

关于前端如何下载后端接口返回response-type为application/octet-stream的文件 问题描述 后端接口定义为直接返回一个文件&#xff0c;如果带认证信息可以直接通过浏览器url下载&#xff0c;但是接口需要传headers认证信息&#xff0c;url上又不支持传相关信息 解决 前端…
阅读更多...
软件工程17-18期末试卷

软件工程17-18期末试卷

2.敏捷开发提倡一个迭代80%以上的时间都在编程&#xff0c;几乎没有设计阶段。敏捷方法可以说是一种无计划性和纪律性的方法。错 敏捷开发是一种软件开发方法论&#xff0c;它强调快速响应变化、持续交付有价值的软件、紧密合作和适应性。虽然敏捷方法鼓励迭代开发和灵活性&…
阅读更多...
好的FPGA编码风格（2）--多参考设计软件的语言模板（Language Templates）

好的FPGA编码风格（2）--多参考设计软件的语言模板（Language Templates）

什么是语言模板&#xff1f; 不论是Xilinx的Vivado&#xff0c;还是Altera的Quartus II&#xff0c;都为开发者提供了一系列Verilog、SystemVerilog、VHDL、TCL、原语、XDC约束等相关的语言模板&#xff08;Language Templates&#xff09;。 在Vivado软件中&#xff0c;按顺序…
阅读更多...
LVS集群-DR模式

LVS集群-DR模式

概念&#xff1a; LVS-DR模式&#xff0c;也是最常用的lVS负载方式&#xff0c;DR DIRECT ROUTING 直接路由模式 负载均衡器lVS调度器&#xff0c;只负责请求和转发到后端的真实服务器&#xff0c;但是影响结果&#xff0c;由后端服务器直接转发给客户端&#xff0c;不需要经…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023