Storm集成Kafka的Trident实现

 

原本打算将storm直接与flume直连，发现相应组件支持比较弱，topology任务对应的supervisor也不一定在哪个节点上，只能采用统一的分布式消息服务Kafka。

 

原本打算将结构设置为：

 

最后结构更改为：

 

  

集成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在数据上执行函数功能、过滤和聚合操作。

 

 

 

 

使用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方法，但是运行时会出现错误：

 

 

 

原因就在于进行适配的过程中，注册方法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创建者编写）

 

https://github.com/nathanmarz/trident-memcached

 

其中调用了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中还存在其他的工具，Storm-contribute项目地址：https://github.com/nathanmarz/storm-contrib，同样是作者所写，加入了支持Redis，Kafka，MongoDB等数据源。