为什么streamparse拓扑定义抱怨thrift$mktopology的参数数目错误？

Question

我试图让一个非常简单的streamparse（即Apache Storm）spout正常工作，但是在运行sparse run -t 120时，我得到了以下错误：

Caught exception: Wrong number of args (1) passed to: thrift$mk-topology

clojure.lang.ArityException: Wrong number of args (1) passed to: thrift$mk-topology
 at clojure.lang.AFn.throwArity (AFn.java:437)
    clojure.lang.AFn.invoke (AFn.java:39)
    clojure.lang.AFn.applyToHelper (AFn.java:161)
    clojure.lang.AFn.applyTo (AFn.java:151)
    clojure.core$apply.invoke (core.clj:617)
    streamparse.commands.run$run_local_BANG_.invoke (run.clj:20)
    streamparse.commands.run$_main.doInvoke (run.clj:79)
    clojure.lang.RestFn.invoke (RestFn.java:930)
    clojure.lang.Var.invoke (Var.java:460)
    user$eval5.invoke (form-init1145748518959444179.clj:1)
    clojure.lang.Compiler.eval (Compiler.java:6619)
    clojure.lang.Compiler.eval (Compiler.java:6609)
    clojure.lang.Compiler.load (Compiler.java:7064)
    clojure.lang.Compiler.loadFile (Compiler.java:7020)
    clojure.main$load_script.invoke (main.clj:294)
    clojure.main$init_opt.invoke (main.clj:299)
    clojure.main$initialize.invoke (main.clj:327)
    clojure.main$null_opt.invoke (main.clj:362)
    clojure.main$main.doInvoke (main.clj:440)
    clojure.lang.RestFn.invoke (RestFn.java:421)
    clojure.lang.Var.invoke (Var.java:419)
    clojure.lang.AFn.applyToHelper (AFn.java:163)
    clojure.lang.Var.applyTo (Var.java:532)
    clojure.main.main (main.java:37)
993  [main] ERROR org.apache.zookeeper.server.NIOServerCnxnFactory - Thread Thread[main,5,main] died
java.lang.NullPointerException: null
    at streamparse.commands.run$run_local_BANG_.invoke(run.clj:33) ~[streamparse-0.0.4-SNAPSHOT.jar:na]
    at streamparse.commands.run$_main.doInvoke(run.clj:79) ~[streamparse-0.0.4-SNAPSHOT.jar:na]
    at clojure.lang.RestFn.invoke(RestFn.java:930) ~[clojure-1.5.1.jar:na]
    at clojure.lang.Var.invoke(Var.java:460) ~[clojure-1.5.1.jar:na]
    at user$eval5.invoke(form-init1145748518959444179.clj:1) ~[na:na]
    at clojure.lang.Compiler.eval(Compiler.java:6619) ~[clojure-1.5.1.jar:na]
    at clojure.lang.Compiler.eval(Compiler.java:6609) ~[clojure-1.5.1.jar:na]
    at clojure.lang.Compiler.load(Compiler.java:7064) ~[clojure-1.5.1.jar:na]
    at clojure.lang.Compiler.loadFile(Compiler.java:7020) ~[clojure-1.5.1.jar:na]
    at clojure.main$load_script.invoke(main.clj:294) ~[clojure-1.5.1.jar:na]
    at clojure.main$init_opt.invoke(main.clj:299) ~[clojure-1.5.1.jar:na]
    at clojure.main$initialize.invoke(main.clj:327) ~[clojure-1.5.1.jar:na]
    at clojure.main$null_opt.invoke(main.clj:362) ~[clojure-1.5.1.jar:na]
    at clojure.main$main.doInvoke(main.clj:440) ~[clojure-1.5.1.jar:na]
    at clojure.lang.RestFn.invoke(RestFn.java:421) ~[clojure-1.5.1.jar:na]
    at clojure.lang.Var.invoke(Var.java:419) ~[clojure-1.5.1.jar:na]
    at clojure.lang.AFn.applyToHelper(AFn.java:163) ~[clojure-1.5.1.jar:na]
    at clojure.lang.Var.applyTo(Var.java:532) ~[clojure-1.5.1.jar:na]
    at clojure.main.main(main.java:37) ~[clojure-1.5.1.jar:na]
Traceback (most recent call last):
  File "/usr/local/bin/sparse", line 9, in <module>
    load_entry_point('streamparse==1.1.0', 'console_scripts', 'sparse')()
  File "/usr/local/lib/python2.7/dist-packages/streamparse/cmdln.py", line 83, in main
    run_local_topology(args["--name"], time, par, options, args["--debug"])
  File "/home/louis/.local/lib/python2.7/site-packages/invoke/tasks.py", line 111, in __call__
    result = self.body(*args, **kwargs)
  File "/usr/local/lib/python2.7/dist-packages/streamparse/ext/invoke.py", line 182, in run_local_topology
    run(full_cmd)
  File "/home/louis/.local/lib/python2.7/site-packages/invoke/runner.py", line 349, in run
    return runner.run(command, **kwargs)
  File "/home/louis/.local/lib/python2.7/site-packages/invoke/runner.py", line 153, in run
    raise Failure(result)
invoke.exceptions.Failure: Command execution failure!

Exit code: 1

Stderr:

我的拓扑.clj文件如下：

^{pr2}$

最后，这是喷口的.py文件。它初始化网络摄像头并开始以numpy数组的形式生成帧：

from __future__ import absolute_import, unicode_literals
import logging

import cv2

from streamparse.spout import Spout


class FrameSpout(Spout):
    def initialize(self, stormconf, context):
        self.cam = cv2.VideoCapture(0)

    def next_tuple(self):
        got_image, img = self.cam.read()
        if not got_image:
            logging.debug("Failed to grab frame")
        else:
            self.emit([img])

我确定我忽略了一些简单的东西，但是由于我对streamparse和Storm比较陌生，所以我不确定应该在哪里查找。任何建议都将不胜感激！在

Answer 1

问题是拓扑定义需要两个映射的列表。一个贴图用于喷口定义，第二个贴图用于螺栓定义。创建仅喷口地形的解决方案是将第二个空映射添加到.clj文件：

(ns facetrack
(:use     [streamparse.specs])
(:gen-class))

(defn facetrack [options]
   [
    ;; spout configuration
    {"frame-spout" (python-spout-spec
          options
          "spouts.frames.FrameSpout"
          ["frame"]
          )
    }
    {}  ;; <   THIS
  ]
)