一 Storm 基础

1 Storm 分布式计算结构称为 Topology （拓扑）

Topology 由 stream（数据流）。spout（数据流的生成者）。bolt（运算）组成。

1.1 stream

Strom 核心数据结构是 tuple。 tuple/英译元组，重数/

1.2 spout

spout 代表了一个 Storm Topology 的数据入口，充当采集器角色。连接到数据源，

并将 tuple 作为数据流进行发射。

{ “sentence”:“My dog is called 'cow cow' ”}

spout 将会一直循环发送 tuple。一般spout 会连接到动态数据源上。

1.3 bolt

可以理解为程序中的运算或函数，将一个或多个数据流作为输入，对数据事实运算后，选择输出一个或者多个数据流。bolt 可以订阅多个 spot 或者其他 bolt 发射的数据流，这样可以建立复杂的数据流转换网络。

bolt 典型功能过滤 tuple 连接和聚合操作。计算。数据库读写。

单词计数 topology 数据流。

1.2.1 语句分割 bolt

语句分割 bolt (SplitSentenceBolt）类将会订阅 sentence spout 发射的 tuple 流，每当受到一个 tuple，bolt 会获取“sentence”对应值域的语句，然后将语句分割为一个个的单词。每个单词向后发射一个 tuple：

1.2.2 单词计数 bolt

单词计数 bolt （WordCountBolt）订阅 SplitSentenceBolt 类的输出，保存每个特定单词出现的次数。每当 bolt 接收到一个 tuple，会将对应单词的计数加一，并且向后发送该单词当前的计数。

1.2.3 上报 bolt

上报 bolt 订阅 WordCountBolt 类的输出，像WordCountBolt 一样，维护一份所有单词对应的计数的表。当接受到一个 tuple 时，上报 bolt 会更新表中的计数数据，并且将值在终端打印。

数据源：数据源继承 BaseRichSpout

实现：实现继承 BaseRichBolt

Topology： WordCountTopology

一个 Strom 程序编写

public class WordCountTopology {

private static final String SENTENCE_SPOUT_ID = "sentence-spout";

private static final String SPLIT_BOLT_ID = "split-bolt";

private static final String COUNT_BOLT_ID = "count-bolt";

private static final String REPORT_BOLT_ID = "report-bolt";

private static final String TOPOLOGY_NAME = "word-count-topology";

public static void main(String[] args) throws Exception {

// 1 首先编写 spout 和 bolt

SentenceSpout spout = new SentenceSpout();

SplitSentenceBolt splitBolt = new SplitSentenceBolt();

WordCountBolt countBolt = new WordCountBolt();

ReportBolt reportBolt = new ReportBolt();

// 2 生成一个 TopologyBuilder 实例

TopologyBuilder builder = new TopologyBuilder();

/** 注册一个 sentence spout 并复制给其唯一的 ID */

// 2 参数为单个。三个参数可以分配多个 task

builder.setSpout(SENTENCE_SPOUT_ID, spout);

/**

* TopologyBuilder 的 setBolt（）方法会注册一个 bolt，并且返回 BoltDeclarer 的实例，可以定义 bolt 的数据源。

* 注册一个 SplitSentenceBolt，这个 bolt 订阅 SentenceSpout发射出来的数据流

builder.setBolt(SPLIT_BOLT_ID, splitBolt).shuffleGrouping(SENTENCE_SPOUT_ID);

builder.setBolt(COUNT_BOLT_ID, countBolt).fieldsGrouping(SPLIT_BOLT_ID, new Fields("word"));

builder.setBolt(REPORT_BOLT_ID, reportBolt).globalGrouping(COUNT_BOLT_ID);

/**

* 所有数据流都应经定义好，运行单词计数计算最后一步名编译并提交到集群上。

* 本地模式。模拟集群。可以设置断电，暂停运行，观察变量，分析程序性能。当 topology 发布到分布式集群后。将很难做到。

* 当一个 Topology 提交时，Storm 会将默认配置和 Config 实例中的配置合并后作为参数传递给 submitTopology()方法。

* 合并后的配置分发给各个 spout 的 bolt 的 open（）、prepare（）方法。

* waitForSeconds（10）停10秒。

* cluster.killTopology(TOPOLOGY_NAME)；停止/卸载该 topology，最后关闭本地模式集群。

* 执行完毕后出 --final counts--

Config config = new Config();

// 以后会用到 config.setNumWorkers(2)；// 分配 Topology 两个 worker

LocalCluster cluster = new LocalCluster();

cluster.submitTopology(TOPOLOGY_NAME, config, builder.createTopology());

Utils.sleep(10000);

cluster.killTopology(TOPOLOGY_NAME);

cluster.shutdown();

}

God has given me a gift. Only one. I am the most complete fighter in the world. My whole life, I have trained. I must prove I am worthy of someting. rocky_24

来自为知笔记(Wiz)

巴特西

一 Storm 基础

最新文章

热门文章