Fix issue where parallelism dicts did not work with sparse run

pystorm · Aug 29, 2017 · bafb72b · bafb72b
1 parent e42803f
commit bafb72b
Showing 1 changed file with 5 additions and 0 deletions.
diff --git a/streamparse/cli/run.py b/streamparse/cli/run.py
@@ -37,6 +37,11 @@ def run_local_topology(name=None, env_name=None, time=0, options=None):
         storm_options['topology.acker.executors'] = 1
     storm_options['topology.workers'] = 1
 
+    # Set parallelism based on env_name if necessary
+    for spec in topology_class.specs:
+        if isinstance(spec.par, dict):
+            spec.par = spec.par.get(env_name)
+
     # Check Storm version is the same
     local_version = local_storm_version()
     project_version = storm_lib_version()