Storm - Conditionally consuming stream from kafka spout? -

-

i have scenario posting json kafka instance. using kafka spout emit stream bolt.

but add additional field (call x) json message. if x a consumed bolta, if x b consumed boltb.

is there way direct stream proper bolt depending on streams contents?

the simplest way should add splitbolt consumes kafkaspout, evaluates field x , , forwards different output streams:

public class splitbolt extends baserichbolt {   outputcollector collector;    public void prepare(...) {     this.collector = collector;   }    public void execute(tuple input) {     object x = ... // field x input     string streamid;     if(x == a) {       streamid = "stream-xa";     } else { // x == b       streamid = "stream-xb";     }     collector.emit(streamid, input, input.getvalues());   }    public void declareoutputfields(outputfieldsdeclarer declarer) {     fields schema = new fields(...)     declarer.declarestream("stream-xa", schema);     declarer.declarestream("stream-xy", schema);   } }

when building topology, connect bolta "stream-xa" , boltb "stream-xb":

topologybuilder b = new topologybuilder(); b.setspout("spout", new kafkaspout(...)); b.setbolt("split", new splitbolt()).shufflegrouping("spout"); b.setbolt("bolta", new bolta()).shufflegrouping("split", "stream-xa"); b.setbolt("boltb", new boltb()).shufflegrouping("split", "stream-xb");

as alternative, should possible, inherit kafkaspout , emit 2 different streams directly. however, code more tricky right.


Comments

Post a Comment

Popular posts from this blog

c# - Store DBContext Log in other EF table -

-
i want store logs in db table (imported ef code). my code looks this: public pmcontext() : base("myentities") { configuration.proxycreationenabled = false; configuration.lazyloadingenabled = false; this.database.log = s => logstore(s); } private void logstore(string message) { tlog log = new tlog(); log.description = message; log.insertdate = datetime.now; logs.add(log); savechanges(); } public virtual dbset<tlog> logs { get; set; } i cannot use savechanges() (get error "sqlconnection not support parallel transactions") , if remove line nothing stored in db table. how can fix this? i think happens during savechanges call anywhere in app, because sql generated call logged, triggers nested savechanges call tries start own transaction. and there thing: savechanges call in logstore method trigger logging itself, you'd end starting infin...
Read more

Nuget pack csproj using nuspec -

-
i want create nuget package contains specified in nuspec file, still version csproj. in order use token have pack like: nuget pack myproj.csproj but when adds dependencies , creates unwanted folder in nuget package. nuspec file is: <?xml version="1.0" encoding="utf-8"?> <package xmlns="http://schemas.microsoft.com/packaging/2011/08/nuspec.xsd"> <metadata> <id>test</id> <version>$version$</version> <title>test</title> <authors>test</authors> <owners>test</owners> <requirelicenseacceptance>false</requirelicenseacceptance> <description>test</description> <summary>test</summary> <releasenotes>test</releasenotes> <copyright>test</copyright> </metadata> <files> <file src="bin\debug\*.dll" target="lib\net45" /> <file s...
Read more

javascript - Karma not able to start PhantomJS on Windows - Error: spawn UNKNOWN -

-
i lost "unknown" error: karma.conf.js: frameworks: ['mocha', 'should'], plugins: ['karma-mocha', 'karma-should', 'karma-phantomjs-launcher'], ... browsers: ['phantomjs'], error: 23 07 2015 14:35:37.691:warn [karma]: no captured browser, open http://localhost:9876/ 23 07 2015 14:35:37.701:info [karma]: karma v0.13.3 server started @ http://localhost:9876/ 23 07 2015 14:35:37.707:info [launcher]: starting browser phantomjs 23 07 2015 14:35:37.738:error [karma]: { [error: spawn unknown] code: 'unknown', errno: 'unknown', syscall: 'spawn' } error: spawn unknown @ exports._errnoexception (util.js:749:11) @ childprocess.spawn (child_process.js:1093:11) @ exports.spawn (child_process.js:933:9) @ object._execcommand (c:\users\jinga4x\git\mdl-sample-project\node_modules\karma\lib\launchers\process.js:63:21) @ object._start (c:\users\jinga4x\git\md...
Read more