trident教程

时间  2019-11-11
标签 trident 教程 繁體版
原文   原文链接

 



（一）理论基础
更多理论之后再补充，或者参考书籍
一、trident是什么？
Trident is a high-level abstraction for doing realtime computing on top of Storm. It allows you to seamlessly intermix high throughput (millions of messages per second), stateful stream processing with low latency distributed querying. If you're familiar with high level batch processing tools like Pig or Cascading, the concepts of Trident will be very familiar – Trident has joins, aggregations, grouping, functions, and filters. In addition to these, Trident adds primitives for doing stateful, incremental processing on top of any database or persistence store. Trident has consistent, exactly-once semantics, so it is easy to reason about Trident topologies.

简单的说，trident是storm的更高层次抽象，相对storm，它主要提供了2个方面的好处：

（1）提供了更高层次的抽象，将经常使用的count,sum等封装成了方法，能够直接调用，不须要本身实现。

（2）提供了一次原语，如groupby等。

（3）提供了事务支持，能够保证数据均处理且只处理了一次。

二、trident每次处理消息均为batch为单位，即一次处理多个元组。


java

三、事务类型docker

关于事务类型，有2个比较容易混淆的概念：spout的事务类型以及事务状态。服务器

它们都有3种类型，分别为：事务型、非事务型和透明事务型。app

（1）spoutless

spout的类型指定了因为下游出现问题致使元组须要重放时，应该怎么发送元组。ide

事务型spout:重放时能保证同一个批次发送同一批元组。能够保证每个元组都被发送且只发送一个，且同一个批次所发送的元组是同样的。函数

非事务型spout：没有任何保障，发完就算。ui

透明事务型spout：同一个批次发送的元组有可能不一样的，它能够保证每个元组都被发送且只发送一次，但不能保证重放时同一个批次的数据是同样的。这对于部分失效的状况尤为有用，假如以kafka做为spout，当一个topic的某个分区失效时，能够用其它分区的数据先造成一个批次发送出去，若是是事务型spout，则必须等待那个分区恢复后才能继续发送。spa

这三种类型能够分别经过实现ITransactionalSpout、ITridentSpout、IOpaquePartitionedTridentSpout接口来定义。.net

 

（2）state

state的类型指定了若是将storm的中间输出或者最终输出持久化到某个地方（如内存），当某个批次的数据重放时应该若是更新状态。state对于下游出现错误的状况尤为有用。

事务型状态：同一批次tuple提供的结果是相同的。

非事务型状态：没有回滚能力，更新操做是永久的。

透明事务型状态：更新操做基于先前的值，这样因为这批数据发生变化，对应的结果也会发生变化。透明事务型状态除了保存当前数据外，还要保存上一批数据，当数据重放时，能够基于上一批数据做更新。




（二）看官方提供的示例

package org.ljh.tridentdemo;

import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.LocalDRPC;
import backtype.storm.StormSubmitter;
import backtype.storm.generated.StormTopology;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;
import storm.trident.TridentState;
import storm.trident.TridentTopology;
import storm.trident.operation.BaseFunction;
import storm.trident.operation.TridentCollector;
import storm.trident.operation.builtin.Count;
import storm.trident.operation.builtin.FilterNull;
import storm.trident.operation.builtin.MapGet;
import storm.trident.operation.builtin.Sum;
import storm.trident.testing.FixedBatchSpout;
import storm.trident.testing.MemoryMapState;
import storm.trident.tuple.TridentTuple;


public class TridentWordCount {
    public static class Split extends BaseFunction {
        @Override
        public void execute(TridentTuple tuple, TridentCollector collector) {
            String sentence = tuple.getString(0);
            for (String word : sentence.split(" ")) {
                collector.emit(new Values(word));
            }
        }
    }

    public static StormTopology buildTopology(LocalDRPC drpc) {
        FixedBatchSpout spout =
                new FixedBatchSpout(new Fields("sentence"), 3, new Values(
                        "the cow jumped over the moon"), new Values(
                        "the man went to the store and bought some candy"), new Values(
                        "four score and seven years ago"),
                        new Values("how many apples can you eat"), new Values(
                                "to be or not to be the person"));
        spout.setCycle(true);

        //建立拓扑对象
        TridentTopology topology = new TridentTopology();
        
        //这个流程用于统计单词数据，结果将被保存在wordCounts中
        TridentState wordCounts =
                topology.newStream("spout1", spout)
                        .parallelismHint(16)
                        .each(new Fields("sentence"), new Split(), new Fields("word"))
                        .groupBy(new Fields("word"))
                        .persistentAggregate(new MemoryMapState.Factory(), new Count(),
                                new Fields("count")).parallelismHint(16);
        //这个流程用于查询上面的统计结果
        topology.newDRPCStream("words", drpc)
                .each(new Fields("args"), new Split(), new Fields("word"))
                .groupBy(new Fields("word"))
                .stateQuery(wordCounts, new Fields("word"), new MapGet(), new Fields("count"))
                .each(new Fields("count"), new FilterNull())
               .aggregate(new Fields("count"), new Sum(), new Fields("sum"));
        return topology.build();
    }

    public static void main(String[] args) throws Exception {
        Config conf = new Config();
        conf.setMaxSpoutPending(20);
        if (args.length == 0) {
            LocalDRPC drpc = new LocalDRPC();
            LocalCluster cluster = new LocalCluster();
            cluster.submitTopology("wordCounter", conf, buildTopology(drpc));
            for (int i = 0; i < 100; i++) {
                System.out.println("DRPC RESULT: " + drpc.execute("words", "cat the dog jumped"));
                Thread.sleep(1000);
            }
        } else {
            conf.setNumWorkers(3);
            StormSubmitter.submitTopologyWithProgressBar(args[0], conf, buildTopology(null));
        }
    }
}





实例实现了最基本的wordcount功能，而后将结果输出。关键步骤以下：



一、定义了输入流

        FixedBatchSpout spout =
                new FixedBatchSpout(new Fields("sentence"), 3, new Values(
                        "the cow jumped over the moon"), new Values(
                        "the man went to the store and bought some candy"), new Values(
                        "four score and seven years ago"),
                        new Values("how many apples can you eat"), new Values(
                                "to be or not to be the person"));
        spout.setCycle(true);



（1）使用FixedBatchSpout建立一个输入spout，spout的输出字段为sentence，每3个元组做为一个batch。
（2）数据不断的重复发送。


二、统计单词数量

        TridentState wordCounts =
                topology.newStream("spout1", spout)
                        .parallelismHint(16)
                        .each(new Fields("sentence"), new Split(), new Fields("word"))
                        .groupBy(new Fields("word"))
                        .persistentAggregate(new MemoryMapState.Factory(), new Count(),
                                new Fields("count")).parallelismHint(16);



这个流程用于统计单词数据，结果将被保存在wordCounts中。6行代码的含义分别为：

（1）首先从spout中读取消息，spout1定义了zookeeper中用于保存这个拓扑的节点名称。

（2）并行度设置为16，即16个线程同时从spout中读取消息。

（3）each中的三个参数分别为：输入字段名称，处理函数，输出字段名称。即从字段名称叫sentence的数据流中读取数据，而后通过new Split()处理后，以word做为字段名发送出去。其中new Split()后面介绍，它的功能就是将输入的内容以空格为界做了切分。

（4）将字段名称为word的数据流做分组，即相同值的放在一组。

（5）将已经分好组的数据做统计，结果放到MemoryMapState，而后以count做为字段名称将结果发送出去。这步骤会同时存储数据及状态，并将返回TridentState对象。

（6）并行度设置。

三、输出统计结果

        topology.newDRPCStream("words", drpc)
                .each(new Fields("args"), new Split(), new Fields("word"))
                .groupBy(new Fields("word"))
                .stateQuery(wordCounts, new Fields("word"), new MapGet(), new Fields("count"))
                .each(new Fields("count"), new FilterNull())
               .aggregate(new Fields("count"), new Sum(), new Fields("sum"));



这个流程从上述的wordCounts对象中读取结果，并返回。6行代码的含义分别为：

（1）等待一个drpc调用，从drpc服务器中接受words的调用来提供消息。调用代码以下：

drpc.execute("words", "cat the dog jumped")
（2）输入为上述调用中提供的参数，通过Split()后，以word做为字段名称发送出去。

（3）以word的值做分组。

（4）从wordCounts对象中查询结果。4个参数分别表明：数据来源，输入数据，内置方法（用于从map中根据key来查找value），输出名称。

（5）过滤掉空的查询结果，如本例中，cat和dog都没有结果。

（6）将结果做统计，并以sum做为字段名称发送出去，这也是DRPC调用所返回的结果。若是没有这一行，最后的输出结果

DRPC RESULT: [["cat the dog jumped","the",2310],["cat the dog jumped","jumped",462]]
加上这一行后，结果为：
DRPC RESULT: [[180]]

四、split的字义

    public static class Split extends BaseFunction {
        @Override
        public void execute(TridentTuple tuple, TridentCollector collector) {
            String sentence = tuple.getString(0);
            for (String word : sentence.split(" ")) {
                collector.emit(new Values(word));
            }
        }
    }


注意它最后会发送数据。

五、建立并启动拓扑

    public static void main(String[] args) throws Exception {
        Config conf = new Config();
        conf.setMaxSpoutPending(20);
        if (args.length == 0) {
            LocalDRPC drpc = new LocalDRPC();
            LocalCluster cluster = new LocalCluster();
            cluster.submitTopology("wordCounter", conf, buildTopology(drpc));
            for (int i = 0; i < 100; i++) {
                System.out.println("DRPC RESULT: " + drpc.execute("words", "cat the dog jumped"));
                Thread.sleep(1000);
            }
        } else {
            conf.setNumWorkers(3);
            StormSubmitter.submitTopologyWithProgressBar(args[0], conf, buildTopology(null));
        }
    }


（1）当无参数运行时，启动一个本地的集群，及自已建立一个drpc对象来输入。
（2）当有参数运行时，设置worker数量为3，而后提交拓扑到集群，并等待远程的drpc调用。


 

（三）使用kafka做为数据源的一个例子

 

 

package com.netease.sytopology;

import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
import java.util.Arrays;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


import storm.kafka.BrokerHosts;
import storm.kafka.StringScheme;
import storm.kafka.ZkHosts;
import storm.kafka.trident.OpaqueTridentKafkaSpout;
import storm.kafka.trident.TridentKafkaConfig;
import storm.trident.TridentTopology;
import storm.trident.operation.BaseFunction;
import storm.trident.operation.TridentCollector;
import storm.trident.operation.builtin.Count;
import storm.trident.testing.MemoryMapState;
import storm.trident.tuple.TridentTuple;
import backtype.storm.Config;
import backtype.storm.StormSubmitter;
import backtype.storm.generated.AlreadyAliveException;
import backtype.storm.generated.InvalidTopologyException;
import backtype.storm.generated.StormTopology;
import backtype.storm.spout.SchemeAsMultiScheme;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;

/*
 * 本类完成如下内容
 */
public class SyTopology {

    public static final Logger LOG = LoggerFactory.getLogger(SyTopology.class);

    private final BrokerHosts brokerHosts;

    public SyTopology(String kafkaZookeeper) {
        brokerHosts = new ZkHosts(kafkaZookeeper);
    }

    public StormTopology buildTopology() {
        TridentKafkaConfig kafkaConfig = new TridentKafkaConfig(brokerHosts, "ma30", "storm");
        kafkaConfig.scheme = new SchemeAsMultiScheme(new StringScheme());
        // TransactionalTridentKafkaSpout kafkaSpout = new
        // TransactionalTridentKafkaSpout(kafkaConfig);
        OpaqueTridentKafkaSpout kafkaSpout = new OpaqueTridentKafkaSpout(kafkaConfig);
        TridentTopology topology = new TridentTopology();

        // TridentState wordCounts =
        topology.newStream("kafka4", kafkaSpout).
        each(new Fields("str"), new Split(),
                new Fields("word")).groupBy(new Fields("word"))
                .persistentAggregate(new MemoryMapState.Factory(), new Count(),
                        new Fields("count")).parallelismHint(16);
        // .persistentAggregate(new HazelCastStateFactory(), new Count(),
        // new Fields("aggregates_words")).parallelismHint(2);


        return topology.build();
    }

    public static void main(String[] args) throws AlreadyAliveException, InvalidTopologyException {
        String kafkaZk = args[0];
        SyTopology topology = new SyTopology(kafkaZk);
        Config config = new Config();
        config.put(Config.TOPOLOGY_TRIDENT_BATCH_EMIT_INTERVAL_MILLIS, 2000);

        String name = args[1];
        String dockerIp = args[2];
        config.setNumWorkers(9);
        config.setMaxTaskParallelism(5);
        config.put(Config.NIMBUS_HOST, dockerIp);
        config.put(Config.NIMBUS_THRIFT_PORT, 6627);
        config.put(Config.STORM_ZOOKEEPER_PORT, 2181);
        config.put(Config.STORM_ZOOKEEPER_SERVERS, Arrays.asList(dockerIp));
        StormSubmitter.submitTopology(name, config, topology.buildTopology());

    }

    static class Split extends BaseFunction {
        public void execute(TridentTuple tuple, TridentCollector collector) {
            String sentence = tuple.getString(0);
            for (String word : sentence.split(",")) {
                try {
                    FileWriter fw = new FileWriter(new File("/home/data/test/ma30/ma30.txt"),true);
                    fw.write(word);
                    fw.flush();
                    fw.close();
                } catch (IOException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }
                collector.emit(new Values(word));
                
            }
        }
    }
}



 

本例将从kafka中读取消息，而后对消息根据“，”做拆分，并写入一个本地文件。
一、定义kafka想着配置
        TridentKafkaConfig kafkaConfig = new TridentKafkaConfig(brokerHosts, "ma30", "storm");
        kafkaConfig.scheme = new SchemeAsMultiScheme(new StringScheme());

        OpaqueTridentKafkaSpout kafkaSpout = new OpaqueTridentKafkaSpout(kafkaConfig);

其中ma30是订阅的topic名称。


二、从kafka中读取消息并处理

        topology.newStream("kafka4", kafkaSpout).
        each(new Fields("str"), new Split(),new Fields("word")).
        groupBy(new Fields("word"))
        .persistentAggregate(new MemoryMapState.Factory(), new Count(),
                        new Fields("count")).parallelismHint(16);

（1）指定了数据来源，并指定zookeeper中用于保存数据的位置，即保存在/transactional/kafka4。
（2）指定处理方法及发射的字段。
（3）根据word做分组。

（4）计数后将状态写入MemoryMapState

 

提交拓扑：

storm jar target/sytopology2-0.0.1-SNAPSHOT.jar com.netease.sytopology.SyTopology 192.168.172.98:2181/kafka test3 192.168.172.98

此时能够在/home/data/test/ma30/ma30.txt看到split的结果

相关文章
相关标签/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公众号
   欢迎关注本站公众号,获取更多信息
联系我们 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程