Storm集成Kafka的Trident实现

时间:2023-01-30 17:21:06
原本打算将storm直接与flume直连,发现相应组件支持比较弱,topology任务对应的supervisor也不一定在哪个节点上,只能采用统一的分布式消息服务Kafka。
原本打算将结构设置为:

Storm集成Kafka的Trident实现
 
最后结构更改为:

Storm集成Kafka的Trident实现
  

集成Kafka

storm中已经写好了KafkaSpout用来接收Kafka中间件上的消息,并发射到Bolt中,只需要依赖 storm-kafka即可:
<dependency>
<groupId>org.apache.storm</groupId>
<artifactId>storm-kafka</artifactId>
<version>${storm.version}</version>
</dependency>
调用org.apache.storm.kafka.KafkaSpout,  需要传递一个SpoutConfig用来配置kafka对应的zookeeper以及topic:
String zks = "192.168.1.1xx:2181,192.168.1.1xx:2181,192.168.1.1xx:2181/kafka";
String topic = "log-storm"; BrokerHosts brokerHosts = new ZkHosts(zks);
SpoutConfig spoutConfig = new SpoutConfig(brokerHosts, topic, "/"+ topic, UUID.randomUUID().toString());
spoutConfig.scheme = new SchemeAsMultiScheme(new StringScheme());
spoutConfig.zkServers = Arrays.asList("192.168.1.1xx","192.168.1.1xx","192.168.1.1xx");
spoutConfig.zkPort = 2181;
建立KafkaSpout(spoutConfig)即可。
需要注意的是,我们在Bolt中需要对收到的消息进行主动ack/fail,否则会出现消息重复发送的情况,一般情况下Bolt的写法类似下面,在prepare中缓存collector,executor中通过try/catch块决定是否确认消息(以通知Spout是否需要对消息进行重发),declareOutputFields中声明需要输出的字段。
private OutputCollector collector;

    @Override
public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
this.collector = collector;
} @Override
public void execute(Tuple input) {
try {
String msgBody = input.getString(0);
int traceIndex = msgBody.indexOf(TRACE_CONST);
if (traceIndex >= 0) {
String completeLog = msgBody.substring(traceIndex + TRACE_CONST.length());
collector.emit(new Values(completeLog));
}
collector.ack(input);
} catch (Exception e) {
collector.fail(input);
}
} @Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {
declarer.declare(new Fields("log"));
}
Storm中需要有一个main函数,用于构建和启动topology,以便将spout,bolt等组件连接起来,代码类似下面:
TopologyBuilder builder = new TopologyBuilder();
builder.setSpout("kafka-reader", new KafkaSpoutNoMetrics(spoutConfig), 3); builder.setBolt("log-extractor", new LogExtractorBolt(), 2).shuffleGrouping("kafka-reader");
builder.setBolt("log-splitter", new LogSplitterBolt(), 2).shuffleGrouping("log-extractor");
builder.setBolt("memcached-store", new MemcachedBolt()).fieldsGrouping("log-splitter", new Fields("md")); Config config = new Config();
String name = "LogStormProcessor"; config.setNumWorkers(1);
StormSubmitter.submitTopologyWithProgressBar(name, config, builder.createTopology());

使用Storm Trident

Trident在Storm上进行了高级抽象,例如事务处理和状态管理的细节,可以让一批tuple进行离散的事务处理,还允许topology在数据上执行函数功能、过滤和聚合操作。

Storm集成Kafka的Trident实现
 
使用Trident,我们需要使用TridentTopology替换原有的TopologyBuilder构造Storm的拓扑图。在Trident中的spout引入了数据批次(batch)的概念,不像Strom中的spout,Trident Spout必须成批地发送tuple。
在Trident中,spout并没有真正发射tuple,而是把这项工作分解给了BatchCoordinator和Emitter方法,Emitter方法负责发送tuple,BatchCoordinator负责管理批次和元数据,Emitter需要依靠元数据来恰当地进行批次的数据重放。
首先,需要根据ITridentSpout新建一个数据流,
 Stream stream = tridentTopology.newStream("event", kafkaSpout);
在使用KafkaSpout作为TridentSpout时,其默认的输出字段名称为str,
Exception in thread "main" java.lang.IllegalArgumentException: Trying to select non-existent field: 'event' from stream containing fields fields: <[str]>
at org.apache.storm.trident.Stream.projectionValidation(Stream.java:853)
at org.apache.storm.trident.Stream.each(Stream.java:320)
at com.zhen.log.processor.trident.Main.main(Main.java:48)
我们原来使用的KafkaSpout,虽然可以将其直接用于newStream方法,但是运行时会出现错误:

Storm集成Kafka的Trident实现
 
原因就在于进行适配的过程中,注册方法registerMetric只能够被宰ISpout::open()方法中被调用,虽然可以进行合理地改造(由于有一些包访问控制权限的相关依赖,新建一个同名package,并将其中的registerMetric方法删除),但是其事务性不能得到保证,在本人测试的过程中,Kafka的消息不能被正常消费,每次重启服务都会读到完整的所有数据。
但是这并不是推荐的用法,storm-kafka中存在另外一个实现:org.apache.storm.kafka.trident.OpaqueTridentKafkaSpout,可以满足要求:
TridentKafkaConfig kafkaConfig = new TridentKafkaConfig(brokerHosts, topic);
OpaqueTridentKafkaSpout kafkaSpout = new OpaqueTridentKafkaSpout(kafkaConfig); TridentTopology tridentTopology = new TridentTopology();
Stream stream = tridentTopology.newStream("event", kafkaSpout);
而使用OpaqueTridentKafkaSpout时,默认的输出名称为“bytes”,其输出格式也并不是String字符串而是byte[],需要根据编码格式将其转换为String。
在Spout编写完成后,就可以加入后续的运算操作,trident处理是通过创建Stream的各种operation并连接来进行处理的,比较常用的两种运算:filter和function,例如我们下面处理流的方式,每次返回Stream都可以继续用来创建新的数据流:
Stream logStream = stream.each(new Fields("bytes"), new LogExtractorFunction(), new Fields("log"))
.each(new Fields("log"), new LogSplitterFunction(), new Fields("logObject"))
.each(new Fields("logObject"), new LogTypeFilter("TRACE"));
在filter中,继承自BaseFilter,唯一的isKeep方法会根据tuple的属性进行相应过滤操作,需要指定对应输入的Field,filter没有额外输出的多余字段。注意filter中不能改变tuple,如果既想要过滤又想添加字段时必须使用function。
在function中,继承自BaseFunction,通过execute方法来对所有的数据增加额外的字段,并不会删除或者变更已有的字段。使用function需要指定多余输出的Fields,function中发射的字段数要与声明的fields字段数据保持一致。
和function比较类似,aggregator允许topology组合tuple,不同的是,它会替换tuple字段和值,有三种聚合器可以被使用:CombinerAggregator,ReducerAggregator和Aggregator。这里,我们使用的是CombinerAggregator。
CombinerAggregator用来将一个集合的tuple组合到一个单独到一个单独的字段中,定义如下:
public interface CombinerAggregator<T> extends Serializable {
T init(TridentTuple tuple);
T combine(T val1, T val2);
T zero();
}
Storm会对每个tuple调用init方法,然后重复调用combiner方法指导一个分片的数据处理完成,传递给combine方法的两个参数是局部聚合的结果,以及调用了init返回的值,如果没有聚合结果,会直接调用zero方法返回一个自定义空值。
聚合一般需要首先对数据流进行groupBy操作后,在GroupedStream流上进行实际操作,一般情况下,首先根据前面的流输出一个用于分组的键值用于groupBy,然后进行persistentAggregate,根据分组将数据归并计算合并结果。
        logStream
.each(new Fields("logObject"), new LogGroupFunction(), new Fields("key")).groupBy(new Fields("key"))
.persistentAggregate(MemcachedState.nonTransactional(servers), new Fields("logObject"), new LogCombinerAggregator(),
new Fields("statistic"))
注意,使用Trident时也是可以分成多个流的,只需要在特定的节点上,保存本地变量,就可以在其上执行多次each,分出多条路径流进行独立处理(也可以对多条输入流进行合并,这里没有使用到这样高级的功能)。
Stream logStream = stream.each(new Fields("bytes"), new LogExtractorFunction(), new Fields("log"))
.each(new Fields("log"), new LogSplitterFunction(), new Fields("logObject"))
.each(new Fields("logObject"), new LogTypeFilter("TRACE")); logStream.each(new Fields("log"), new LocalFileSaveFunction(), new Fields()); logStream
.each(new Fields("logObject"), new LogGroupFunction(), new Fields("key")).groupBy(new Fields("key"))
.persistentAggregate(MemcachedState.nonTransactional(servers), new Fields("logObject"), new LogCombinerAggregator(),
new Fields("statistic"))
;
在使用任何function,aggregator时,都可以通过声明Fields的方式来设置使用到的字段名称,Combiner中可以不使用任何定义的Fields,此时传递给Trident的Tuple中将不会包含任何字段(一般代码示例中如此)。使用聚合时,还需要持续存储聚合的Trident状态,持久化操作从状态管理开始,Trident对状态有底层的操作原语,可以参考State接口的方法。
Storm中用State来持久化存储信息,有三种状态类型:非事务型,重复事务型以及不透明事务型,在分布式环境下,数据可能被重放,为了支持计数和状态更新,Trident将状态更新操作进行序列化,使用不同的状态更新模式对重放和错误数据进行容错。
我们存储中间数据状态使用了memcached作为媒介,关于trident与memcached进行事务处理相关代码,可以参考工程(storm创建者编写)
其中调用了twitter中定义的所以使用改造过的客户端:
<dependency>
<groupId>com.twitter</groupId>
<artifactId>finagle-memcached_2.9.2</artifactId>
<version>6.20.0</version>
</dependency>
但是将源码copy到工程中并添加对应的maven依赖(多数是twitter相关的依赖),其中的twitter依赖始终找不到:
[INFO] ------------------------------------------------------------------------
[ERROR] Failed to execute goal on project log-storm-processor: Could not resolve dependencies for project com.zhen:log-storm-processor:jar:1.0.0-SNAPSHOT: The following artifacts could not be resolved: com.twitter.common.zookeeper:server-set:jar:1.0.83, com.twitter.common.zookeeper:client:jar:0.0.60, com.twitter.common.zookeeper:group:jar:0.0.78: Failure to find com.twitter.common.zookeeper:server-set:jar:1.0.83 in http://192.168.1.14:8081/nexus/content/repositories/releases/ was cached in the local repository, resolution will not be reattempted until the update interval of nexus-releases has elapsed or updates are forced -> [Help 1]
[ERROR]
[ERROR] To see the full stack trace of the errors, re-run Maven with the -e switch.
因此只能将其进行改造,使用com.whalin对应的memcached客户端jar包,以满足从Storm存储到memcached的需求。
storm trident的结构:

Storm集成Kafka的Trident实现
 
storm中还存在其他的工具,Storm-contribute项目地址:https://github.com/nathanmarz/storm-contrib,同样是作者所写,加入了支持Redis,Kafka,MongoDB等数据源。