小编Mat*_*Sax的帖子

我一直在努力记录当前项目的所有Storm指标的含义.

在此过程中,我从该组响应以及github收集了数据.

虽然有些指标非常自我解释,但我对一些螺栓指标感到困惑.

例如,Process Latency和Execute Latency之间有什么区别？

通过此Google论坛上的帖子,我收集了以下信息:

• 列表项处理延迟=调用ack时的时间戳 - 执行传递元组时的时间戳

• 列表项执行延迟=执行函数结束时的时间戳 - 执行传递时的时间戳元组(来源:http://goo.gl/3KRAl)

和

• 列表项处理延迟是直到元组被激活的时间,执行延迟是元组执行所花费的时间(来源:http://goo.gl/m0fTC)

根据我在风暴用户界面中看到的内容,我的执行延迟几乎总是大于进程延迟.怎么会这样？任何人都可以帮我确定两种延迟的确切定义吗？

提前致谢!

在卡夫卡和风暴背景下构建多租户应用程序的最佳实践是什么？

例如:为每个租户创建主题并使用多主题spout(使用通配符).

我正在尝试编写执行以下操作的拓扑:

1. 订阅Twitter订阅源的订阅源(基于关键字)
2. 聚合螺栓,聚合集合中的许多推文(比如N)并将它们发送给打印机螺栓
3. 一个简单的螺栓,可以立即将集合打印到控制台.

实际上我想对集合进行更多处理.

我在本地测试它,看起来它正在工作.但是,我不确定我是否正确设置了螺栓上的分组,以及在部署在实际风暴群上时这是否可以正常工作.如果有人可以帮助审查此拓扑并建议任何错误,更改或改进,我将不胜感激.

谢谢.

这就是我的拓扑结构.

builder.setSpout("spout", new TwitterFilterSpout("pittsburgh"));
   builder.setBolt("sampleaggregate", new SampleAggregatorBolt())
                .shuffleGrouping("spout");
   builder.setBolt("printaggreator",new PrinterBolt()).shuffleGrouping("sampleaggregate");

聚合螺栓

public class SampleAggregatorBolt implements IRichBolt {

    protected OutputCollector collector;
    protected Tuple currentTuple;
    protected Logger log;
    /**
     * Holds the messages in the bolt till you are ready to send them out
     */
    protected List<Status> statusCache;

    @Override
    public void prepare(Map stormConf, TopologyContext context,
                        OutputCollector collector) {
        this.collector = collector;

        log = Logger.getLogger(getClass().getName());
        statusCache = new ArrayList<Status>();
    }

    @Override
    public void execute(Tuple tuple) { …

我正在研究Spark(伯克利)集群计算系统.在我的研究中,我了解了其他一些内存系统,如Redis,Memcachedb等.如果有人能给我SPARK和REDIS(以及MEMCACHEDB)之间的比较,那将会很棒.在什么情况下Spark比其他内存系统有优势？

我最近熟悉了Apache Kafka,并有一个生产者 - 消费者的工作示例.

我的下一步是将Kafka与Spout和Bolt集成,我很难在本地工作的示例(他们大多是旧的).

我得到以下示例工作storm-b​​ook/examples-ch02-getting_started,它从本地文本文件中读取数据.

同样的回购有一个风暴书/例子-ch04-spouts kafka- spout的例子 但是我无法让它工作.

我尝试了以下示例以及cep.kafka,但得到以下错误 -

5034 [Thread-11-words] INFO  org.apache.curator.framework.imps.CuratorFrameworkImpl - Starting
5047 [Thread-11-words] ERROR backtype.storm.util - Async loop died!
java.lang.NoSuchMethodError: org.apache.zookeeper.ZooKeeper.<init>(Ljava/lang/String;ILorg/apache/zookeeper/Watcher;Z)V
        at org.apache.curator.utils.DefaultZookeeperFactory.newZooKeeper(DefaultZookeeperFactory.java:29) ~[curator-client-2.4.0.jar:na]
        at org.apache.curator.framework.imps.CuratorFrameworkImpl$2.newZooKeeper(CuratorFrameworkImpl.java:169) ~[curator-framework-2.4.0.jar:na]
        at org.apache.curator.HandleHolder$1.getZooKeeper(HandleHolder.java:94) ~[curator-client-2.4.0.jar:na]
        at org.apache.curator.HandleHolder.getZooKeeper(HandleHolder.java:55) ~[curator-client-2.4.0.jar:na]
        at org.apache.curator.ConnectionState.reset(ConnectionState.java:219) ~[curator-client-2.4.0.jar:na]
        at org.apache.curator.ConnectionState.start(ConnectionState.java:103) ~[curator-client-2.4.0.jar:na]
        at org.apache.curator.CuratorZookeeperClient.start(CuratorZookeeperClient.java:188) ~[curator-client-2.4.0.jar:na]
        at org.apache.curator.framework.imps.CuratorFrameworkImpl.start(CuratorFrameworkImpl.java:234) ~[curator-framework-2.4.0.jar:na]
        at storm.kafka.ZkState.<init>(ZkState.java:62) ~[storm-kafka-0.9.2-incubating.jar:0.9.2-incubating]
        at storm.kafka.KafkaSpout.open(KafkaSpout.java:85) ~[storm-kafka-0.9.2-incubating.jar:0.9.2-incubating]
        at backtype.storm.daemon.executor$fn__3371$fn__3386.invoke(executor.clj:522) ~[storm-core-0.9.4.jar:0.9.4]
        at backtype.storm.util$async_loop$fn__460.invoke(util.clj:461) ~[storm-core-0.9.4.jar:0.9.4]
        at clojure.lang.AFn.run(AFn.java:24) [clojure-1.5.1.jar:na]
        at java.lang.Thread.run(Thread.java:745) [na:1.8.0_05]
5049 [Thread-11-words] ERROR backtype.storm.daemon.executor -
java.lang.NoSuchMethodError: org.apache.zookeeper.ZooKeeper.<init>(Ljava/lang/String;ILorg/apache/zookeeper/Watcher;Z)V
        at …

卡住了一下CoFlatMapFunction.如果我把它放在DataStream之前的窗口上似乎工作正常但是如果放在窗口的"应用"功能之后就失败了.

我正在测试两个流,主要的"功能"在flatMap1不断摄取数据和控制流"模型"时flatMap2根据请求更改模型.

我能够设置并看到b0/b1正确设置flatMap2,但flatMap1始终看到b0和b1在初始化时设置为0.

我错过了一些明显的东西吗？

public static class applyModel implements CoFlatMapFunction<Features, Model, EnrichedFeatures> {
    private static final long serialVersionUID = 1L;

    Double b0;
    Double b1;

    public applyModel(){
        b0=0.0;
        b1=0.0;
    }

    @Override
    public void flatMap1(Features value, Collector<EnrichedFeatures> out) {
        System.out.print("Main: " + this + "\n");
    }

    @Override
    public void flatMap2(Model value, Collector<EnrichedFeatures> out) {
        System.out.print("Old Model: " + this + "\n");
        b0 = value.getB0();
        b1 = value.getB1();
        System.out.print("New Model: " …

我们正在评估为部署而追求Storm,但我有点担心.我们目前运行Hadoop MapReduce,并希望将我们的一些处理从MapReduce转换为Storm进程.请注意,这是一些,但不是全部.我们仍然会有一些MapReduce功能.

我找到了Mesos,它可能(可能)允许我们在同一硬件上维护Storm和Hadoop部署,但还有一些其他问题:

• 我设想理想的情况是能够任意地"借用"Storm和Hadoop之间的插槽.恩.两者都会根据需要使用相同的资源.不幸的是,这是一个固定的部署,并不像EC2那样"基于云".

• 我想避免在我们的Storm环境中遇到瓶颈.一个理想的情况是根据需求"旋转"(或反向)螺栓的更多实例.这可能/现实吗？

• "重新启动"拓扑似乎是一个相当昂贵的操作,我不确定是否真的是一个选项.理想情况下,我希望它尽可能无缝.

我们正确接近这个问题吗？从本质上讲,Storm拓扑将"提供"MapReduce批处理作业.我们的一些处理可以以流方式处理,并且作为Storm拓扑会更好,而其中一些需要批处理.

任何一般性反馈,即使它没有解决我的具体问题,也会受到欢迎.在这一点上,这更像是一个探索阶段,我可能完全以错误的方式接近这一点.

我刚开始使用Apache Storm.我阅读了教程并查看了示例我的问题是所有示例都使用非常简单的元组(通常一个字符串存档).元组是内联创建的(使用新值(...)).在我的情况下,我有许多字段的元组(5..100).所以我的问题是如何为每个字段实现名称和类型(所有原始)的这样的元组？

有什么例子吗？(我认为直接实施"元组"并不是一个好主意)

谢谢

我正在评估 AWS Kinesis 与托管服务 Kafka (MSK)。我们的要求是从本地系统（使用 C++ 开发的系统）向 AWS 发送一些消息 (JSON)。然后我们需要将上面的消息持久化到像 PostgreSQL 这样的关系数据库中，同时我们需要将上面的数据流式传输到托管在 AWS 中的其他一些微服务（java）中。

我有以下疑问：

i) 如何从我的本地系统访问（连接和发送消息）到 AWS Kinesis？是否有任何 C++ API 支持？（有 java 客户端 API，但我们的本地系统是用 C++ 编写的）

ii) 如何从我的本地系统访问（连接和发送消息）到 AWS MSK？

iii) 是否可以将 MSK 与其他 AWS 服务（例如 lambda、Redshift、EMR 等）集成？

iv) 要将数据保存到数据库中，我们可以使用 AWS lambda 吗？（AWS Kinesis 支持该功能，AWS MSK 怎么样）

v) 我们的消息速率是 50msg/秒，什么是具有成本效益的解决方案？

我们正在尝试使用Apache Storm处理大量(假)消息.消息示例:

"{"clientName":"Sergey Bakulin","sum":12925,"group":"property","suspicious":false,"clientId":2,"dt":1404387303764,"coord":{"lat":55.767842588357645,"lon":37.46920361823332}}".

我们使用Apache Kafka作为Storm集群的消息来源.我们的目的是能够处理至少50k msg/sec/node.如果我们使用多个节点,我们会不断遇到错误(日志片段来自worker - *.log):

2014-07-03 15:14:47 b.s.m.n.Client [INFO] failed to send requests to ip-172-31-23-123.eu-west-1.compute.internal/172.31.23.123:6701: java.nio.channels.ClosedChannelException: null
at org.jboss.netty.channel.socket.nio.AbstractNioWorker.cleanUpWriteBuffer(AbstractNioWorker.java:381) [netty-3.6.3.Final.jar:na]
at org.jboss.netty.channel.socket.nio.AbstractNioWorker.close(AbstractNioWorker.java:349) [netty-3.6.3.Final.jar:na]
at org.jboss.netty.channel.socket.nio.NioWorker.read(NioWorker.java:93) [netty-3.6.3.Final.jar:na]
at org.jboss.netty.channel.socket.nio.AbstractNioWorker.process(AbstractNioWorker.java:107) [netty-3.6.3.Final.jar:na]
at org.jboss.netty.channel.socket.nio.AbstractNioSelector.run(AbstractNioSelector.java:312) [netty-3.6.3.Final.jar:na]
at org.jboss.netty.channel.socket.nio.AbstractNioWorker.run(AbstractNioWorker.java:88) [netty-3.6.3.Final.jar:na]
at org.jboss.netty.channel.socket.nio.NioWorker.run(NioWorker.java:178) [netty-3.6.3.Final.jar:na]
at org.jboss.netty.util.ThreadRenamingRunnable.run(ThreadRenamingRunnable.java:108) [netty-3.6.3.Final.jar:na]
at org.jboss.netty.util.internal.DeadLockProofWorker$1.run(DeadLockProofWorker.java:42) [netty-3.6.3.Final.jar:na]
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145) [na:1.7.0_51]
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615) [na:1.7.0_51]
at java.lang.Thread.run(Thread.java:744) [na:1.7.0_51]
2014-07-03 15:14:47 b.s.m.n.StormClientErrorHandler [INFO] Connection failed Netty-Client-ip-172-31-23-123.eu-west-1.compute.internal/172.31.23.123:6701

我们当前的风暴配置:

########### These MUST be filled in for a storm configuration
storm.zookeeper.servers:
  - "172.31.*.*"

storm.local.dir: "/home/*/storm/data" …