Storm知识点

Storm知识点

1. 离线计算是什么?

离线计算:批量获取数据、批量传输数据、周期性批量计算数据、数据展示
代表技术:Sqoop批量导入数据、HDFS批量存储数据、MapReduce批量计算数据、Hive批量计算数据、Azkaban任务调度。

2. 流式计算是什么?

流式计算:数据实时产生、数据实时传输、数据实时计算、实时展示
代表技术:Flume实时获取数据、Kafka/metaq实时数据存储、Storm/JStorm实时数据计算、Redis实时结果缓存、持久化存储(mysql)。

3. storm核心组件和架构?

  • 核心组件
DataSource:外部数据源;
Spout:接受外部数据源的组件,以Tuple为基本单元发给Bolt;
Bolt:根据业务逻辑进行处理,发送给下一个Bolt或者是存储到某种介质上,比如Kafka、Redis或者Mysql;
Tuple:数据传输的基本单元,里面封装了一个List对象来保存数据;
  • 架构
Nimbus:任务分配
Supervisor:接受任务,并启动worker。worker的数量根据端口号来的;
Worker:执行任务的具体组件(就是一个JVM),可以执行Spout任务或者bolt任务,一个topology要求的worker数量如果不被满足,在任务分配时根据现有的worker先运行topology,如果当前集群中worker数量为0,那么最新提交的topology将只会被标识active,不会运行,只有当集群有了空闲资源之后,才会被运行;
Task:默认Task=线程=executor。 一个Task属于一个Spout或者Bolt并发任务;
Zookeeper:保存任务分配的信息、心跳信息、元数据信息。
Storm知识点

4. 并发度

用户指定的一个任务,可以被多个线程执行,并发度的数量等于线程的数量。一个任务的多个线程,会被运行在多个Worker(JVM)上。

5. Storm运行模式

本地模式:在当前IDE运行;
集群模式:在集群上运行。

6. Storm常见的分组策略

共有7种,这里记录常见的几种:
shuffleGrouping-随机分组:Random函数;
Non Grouping-不分组:Random函数;
FieldGrouping-字段分组:Hash取模,保证同一类型数据数据流向同一个worker上的task;
Local or ShuffleGrouping-本地随机分组:本地或随机,优先本地。

7. Storm操作命令

  • 任务提交命令:storm jar 【jar路径】 【拓扑包名.拓扑类名】 【拓扑名称】
 storm jar /export/servers/storm/examples/storm-starter/storm-starter-topologies-1.0.3.jar org.apache.storm.starter.WordCountTopology  wordcount
与hadoop不同的是:不需要指定输入输出路径,因为在代码中已经记录了
 hadoop jar /usr/local/wordcount.jar /data.txt /wcout 
  • 杀死任务命令:storm kill 【拓扑名称】 -w 10(执行kill命令时可以通过-w [等待秒数]指定拓扑停用以后的等待时间)
 storm kill topology-name -w 10
  • 停用任务命令:storm deactive  【拓扑名称】
 storm deactive topology-name
我们能够挂起或停用运行中的拓扑。当停用拓扑时,所有已分发的元组都会得到处理,但是spouts的nextTuple方法不会被调用。销毁一个拓扑,可以使用kill命令。它会以一种安全的方式销毁一个拓扑,首先停用拓扑,在等待拓扑消息的时间段内允许拓扑完成当前的数据流。
  • 启用任务命令:storm activate 【拓扑名称】
 storm activate topology-name
  • 重新部署任务命令:storm rebalance  【拓扑名称】
 storm rebalance topology-name
再平衡使你重分配集群任务。这是个很强大的命令。比如,你向一个运行中的集群增加了节点。再平衡命令将会停用拓扑,然后在相应超时时间之后重分配worker,并重启拓扑。
8. Storm Ack机制
为了保证数据能正确的被处理, 对spout产生的每一个tuple, storm都会进行跟踪。即ACK机制,storm启动acker task参与工作流程:
  • Spout创建一个新的Tuple时,会发一个消息通知acker去跟踪;
  • Bolt在处理Tuple成功或失败后,也会发一个消息通知acker;
  • acker会找到发射该Tuple的Spout,回调其ack或fail方法。
Storm的Bolt有BaseBasicBolt和BaseRichBolt,可以在这两种bolt中实现可靠消息传递:
在BaseBasicBolt中,BasicOutputCollector在emit数据的时候,会自动和输入的tuple相关联,而在execute方法结束的时候那个输入tuple会被自动ack。 
 //BaseBasicBolt 实现可靠消息传递
public class SplitSentence extends BaseBasicBolt {//自动建立 anchor,自动 ack
  public void execute(Tuple tuple, BasicOutputCollector collector) {
    String sentence = tuple.getString(0);
    for(String word: sentence.split(" ")) {
      collector.emit(new Values(word));
    }
  }
  public void declareOutputFields(OutputFieldsDeclarer declarer) {
    declarer.declare(new Fields("word"));
  }
}
在使用BaseRichBolt需要在emit数据的时候,显示指定该数据的源tuple要加上参数anchor tuple,以保持tracker链路,即collector.emit(oldTuple, newTuple);并且需要在execute执行成功后调用OutputCollector.ack(tuple), 当失败处理时,执行OutputCollector.fail(tuple);
 //BaseRichBolt 实现可靠消息传递
public class SplitSentence extends BaseRichBolt {//建立 anchor 树以及手动 ack 的例子
OutputCollector _collector;
public void prepare(Map conf, TopologyContext context, OutputCollector collector) {
_collector = collector;
}
public void execute(Tuple tuple) {
String sentence = tuple.getString(0);
for(String word: sentence.split(" ")) {
_collector.emit(tuple, new Values(word)); // 建立 anchor 树
}
_collector.ack(tuple); //手动 ack,如果想让 Spout 重发该 Tuple,则调用 _collector.fail(tuple);
}
public void declareOutputFields(OutputFieldsDeclarer declarer) {
declarer.declare(new Fields("word"));
}
}

9. 调整可靠性

有时候需要调整Storm的可靠性。例如,不关心数据是否丢失,或者想看看后面是否有某个Bolt拖慢Spout速度。有三种方法可以实现:
  • 在build topology时,设置acker数目为0,即conf.setNumAckers(0);
  • 在Spout中,不指定messageId,使得Storm无法追踪;
  • 在Bolt中,使用Unanchor方式发射新的Tuple。

10. Jstorm查看CPU核数

可以用cpu.used / 100向上取整大致估计

Storm知识点

11. Jstorm消费kakfa单条数据太大起Lag

可以设置kafka spout参数:
Kafka spoutConfig.fetchSizeBytes = ;
Kafka spoutConfig.bufferSizeBytes = ;
上一篇:Linux Mysql基础操作


下一篇:初识Linux 基础操作