《OD学storm》20160827

http://www.cnblogs.com/lujinhong2/p/4686512.html

http://blog.csdn.net/paul_wei2008/article/details/20830329

http://shiyanjun.cn/archives/1472.html

一、topology拓扑图

Storm在集群上运行一个Topology时,主要通过以下3个实体来完成Topology的执行工作:
(1)Worker(进程)
(2)Executor(线程)
(3)Task

1个worker进程执行的是1个topology的子集(注:不会出现1个worker为多个topology服务)。1个worker进程会启动1个或多个executor线程来执行1个topology的component(spout或bolt)。因此,1个运行中的topology就是由集群中多台物理机上的多个worker进程组成的。

executor是1个被worker进程启动的单独线程。每个executor只会运行1个topology的1个component(spout或bolt)的task(注:task可以是1个或多个,storm默认是1个component只生成1个task,executor线程里会在每次循环里顺序调用所有task实例)。

task是最终运行spout或bolt中代码的单元(注:1个task即为spout或bolt的1个实例,executor线程在执行期间会调用该task的nextTuple或execute方法)。topology启动后,1个component(spout或bolt)的task数目是固定不变的,但该component使用的executor线程数可以动态调整(例如:1个executor线程可以执行该component的1个或多个task实例)。这意味着,对于1个component存在这样的条件:#threads<=#tasks(即:线程数小于等于task数目)。默认情况下task的数目等于executor线程数目,即1个executor线程只运行1个task。

5. 并发

1)worker工作进程级别的并发设置

2)executor线程级别的并发

真正能够提高Topology性能的并发级别

3)Task任务级别的并发

builder.setBolt(BOLT, BOLT, 2).setNumberTasks(4)

指定2个executor跑4个task。

即每个executor执行2个task,这两个task并不是并发执行,而是轮流执行。

6. 消息可靠性机制

1)Bolt消息可靠性机制

每处理完接收到Tuple,发送确认信息

2)tuple锚定

二、Trident

1. 概念

1)Storm高层次的抽象

2)在Trident中保留了Spout,但是不再有Bolt组件。封装成了一系列的Operation,比如过滤、函数、分组等

3)Trident封装好了消息可靠性保障机制

4)Trident批次概念

将固定条数的Tuple划分为一个批次

给每个批次一个编号

更新统计结果状态,要严格按照批次顺序进行更新

5)事务控制

3个层次:

(1)NON-Transactional:非事务控制

允许同一个批次内的Tuple部分处理成功,失败的Tuple,可以在其他批次内进行重试,也有可能不进行重试。

(2)Transactional:严格的事务控制

要求批次内处理失败的Tuple,只能在本批次内进行重试。

如果tuple一直重试不成功,就会将整个任务程序挂起,不会进行下个批次的处理。没有容错。

(3)Opaque-Transactional:透明事务处理

批次内的tuple处理完成过后,先把成功共的更新掉,失败的tuple允许在其他批次内进行重试,只会有一次成功处理。有容错。

2. Trident的编码开发

1)构造Topology

trident中的Spout:

从数据源上获取数据,将获取到的数据封装到一个批次,并给每个批次执行一个批次号。

2)each方法

过滤操作Filter

对满足条件的Filter,

isKeep方法

3)Filter

在Trident编码中,要注意的:

Stream流在经过各种操作后,Tuple的演变,Tuple中keyvalue对的演变。

比如:Filter

仅仅只是在Stream保留或者丢弃tuple,而不会对tuple进行改变

4)Function

Stream经过函数操作后,将新产生的keyvalue对追加到原来的Tuple中,

需要注意的是,如果没有新产生keyvalue对,那么相应的tuple将会被丢弃掉。

5)指定保留哪些keyvalue、丢弃哪些keyvalue

.project

3. 并发

3)Operation的特性

是否同一分区、是否需要跨网络。

分区:指的就是一个在executor线程中运行的task。

Filter Function project

分区内的操作,只是对本分区内的Tuple进行操作

4)重分区操作 数据流分组

partitionBy ——分区

.partitionBy(new Fields("word"))

取Tuple,key名称为word的keyvalue的value值,求hashcode,然后根据哈希值 % 分区数进行取模。

相同的keyvalue对,进入同一个分区。

同一个分区内keyvalue对是否都相同?不一定

groupBy ——分区加分组

例如 .groupBy(new Fields("word"))

实际进行了两步操作:

(1)partitionBy

(2)在将同一个分区内相同的keyvalue,分配到到一个组。

 启动DRPC server服务

drpc.srevers:

  hostname

drpc.port:

  273

先说一下场景,日志文件中有大约三千万行数据,大小为1.2G,格式为IP,TIME,现在要针对IP字段进行数量统计重复的次数,以便制定规则来控制用户的恶意注册。

shell版【15分钟统计完成】

date
cat regIp.txt | awk -F' ' '{ print $1}' | sort | uniq -c | sort -rn | head -n100
date

原文地址:https://www.cnblogs.com/yeahwell/p/5813026.html