Why is my streamparse topology definition complaining about a wrong number of arguments to thrift$mk-topology?

Question

I'm trying to get a very simple streamparse (i.e. Apache Storm) spout working, but I'm getting the error below when running 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:

My topology .clj file is as follows:

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

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

And lastly, here's the spout's .py file. It initalizes the webcam and begins generating frames in the form of numpy arrays:

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])

I'm sure I'm overlooking something simple, but since I'm rather new to streamparse and Storm, I'm not sure where to look. Any advice would be greatly appreciated!

Answer 1

The issue is that a topology definition needs a list of two maps. One map is for spout definitions, the second map is for bolt definitions. The solution to creating a spout-only topography is to add a second empty map to the .clj file:

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

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