pyspark - How to do a Running (Streaming) reduceByKey in Spark Streaming -


i'm using textfilestream() method in python api spark streaming read in xml files they're created, map them xml elementtree, take "interesting" items elementtree , flatmap them dictionary (key: value), reducebykey() aggregate counts each key.

so, if key string network name, value might packet count. upon reducing, i'm left total packet count each network (key) in dictionary.

my problem i'm having trouble streaming this. instead of keeping running total re-computes computation each time. think it's paradigmatic issue me i'm wondering if can please me stream analytic correctly, thanks!

ah, solution use updatestatebykey doc allows merge results previous step data in current step. in other words, allows keep running calculation without having store entire rdd , having recompute every time data received.


Comments

Post a Comment

Popular posts from this blog

android - IllegalStateException: Cannot call this method while RecyclerView is computing a layout or scrolling -

i trying remove item recyclerview , error. java.lang.illegalstateexception: cannot call method while recyclerview computing layout or scrolling i using notifydatasetchanged() . how can solve this? here adapter code public class listadapters extends recyclerview.adapter<listadapters.myviewholder> { public arraylist<string> tvdatalist; context c; int pos; listadapters.myviewholder myviewholder; private layoutinflater layoutinflater; int[] arr; public class myviewholder extends recyclerview.viewholder implements view.onclicklistener { public edittext edttxt; public checkbox cb; public myviewholder(view view) { super(view); edttxt = (edittext) view.findviewbyid(r.id.edttxt); cb = (checkbox) view.findviewbyid(r.id.cb); } @override public void onclick(view v) { } } public listadapters(context c, arraylist<string> tvdatali...
Read more

c# - ASP.NET Core - There is already an object named 'AspNetRoles' in the database -

i'm receiving error applicationdbcontext database when trying update entity framework. i've googled, tried fixes here on so, no avail. i'm deleting database each time make change, once deploy, not option. here's error: pm> update-database -context applicationdbcontext system.data.sqlclient.sqlexception: there object named 'aspnetroles' in database. @ system.data.sqlclient.sqlconnection.onerror(sqlexception exception, boolean breakconnection, action 1 wrapcloseinaction) @ system.data.sqlclient.tdsparser.throwexceptionandwarning(tdsparserstateobject stateobj, boolean callerhasconnectionlock, boolean asyncclose) @ system.data.sqlclient.tdsparser.tryrun(runbehavior runbehavior, sqlcommand cmdhandler, sqldatareader datastream, bulkcopysimpleresultset bulkcopyhandler, tdsparserstateobject stateobj, boolean& dataready) @ system.data.sqlclient.sqlcommand.runexecutenonquerytds(string methodname, boolean async, int32 timeout, boolean...
Read more

c# - Oracle Advanced Queues - Dequeueing Commit/Rollback -

we're attempting use oracle aq build queueing system our app in .net 4.7. basically, our problem wrap dequeueing process in upper level transaction containing other instructions , able commit or rollback "manually" after executing queue.dequeue() instruction. so far, works enqueueing: using (var tr = con.begintransaction()) { try { enqmsg.senderid = new oracleaqagent("subscriber1"); enqmsg.payload = new oraclexmltype(con, new xdocument( new xelement("workflowexecution", new xelement("id", i), new xelement("workflowname", guid.newguid().tostring().substring(0, 8)), new xelement("requestsource", guid.newguid().tostring().substring(0, 6)))).tostring()); queue.enqueue(enqmsg); //other instructions here... tr.commit(); } catch (exception) { tr.rollback(); } } with same approach, tryi...
Read more