apache kafka - Spark Streaming job behaves like a batch -

-

i'm trying run spark streaming job read messages kafka. testing insert kafka couple of gigabytes of messages , launch streaming job (python).

from spark import sparkcontext, sparkconf pyspark.streaming import streamingcontext  conf = sparkconf() conf.setmaster("spark://master1:7077,master2:7077,master3:7077") sc = sparkcontext(conf=conf) ssc = streamingcontext(sc, 1)  pyspark.streaming.kafka import kafkautils directkafkastream = kafkautils.createdirectstream(ssc, ["test_topic"], {"metadata.broker.list": "kafka1:9092,kafka2:9092,kafka3:9092", "auto.offset.reset": "smallest"})  directkafkastream.count().pprint()  ssc.start() ssc.awaittermination()

i expect job update me every second messages counted far (streaming). job counts messages in queue (whatever time takes) , after messages counted, see update (batch).

please, tell me have missed here.

update: expecting see:

------------------------------------------- time: 2016-01-28 16:23:42 ------------------------------------------- 1000000  ------------------------------------------- time: 2016-01-28 16:23:43 ------------------------------------------- 10000000

what i'm seeing:

16/01/28 16:25:10 info jobscheduler: added jobs time 1453994710000 ms 16/01/28 16:25:11 info jobscheduler: added jobs time 1453994711000 ms 16/01/28 16:25:12 info jobscheduler: added jobs time 1453994712000 ms 16/01/28 16:25:13 info jobscheduler: added jobs time 1453994713000 ms 16/01/28 16:25:14 info jobscheduler: added jobs time 1453994714000 ms 16/01/28 16:25:15 info jobscheduler: added jobs time 1453994715000 ms 16/01/28 16:25:16 info jobscheduler: added jobs time 1453994716000 ms ...


Comments

Post a Comment

Popular posts from this blog

Hatching array of circles in AutoCAD using c# -

-
i beginner in autocad plugins .i have created array of circle using following code. now, have hatch circles in array .how that??how hatch circles @ once? moreover, how can return hatch method , use hatch objects.sorry language.thanks in advance. [assembly: commandclass(typeof(beamsection.class1))] namespace beamsection { public class class1 { //class1 obj = new class1(); [commandmethod("beamatsupport")] public void addlightweightpolyline() { // current document , database document acdoc = application.documentmanager.mdiactivedocument; database accurdb = acdoc.database; promptpointresult ppr = acdoc.editor.getpoint("\nselect starting point "); var ucs = acdoc.editor.currentusercoordinatesystem; point3d startingpt = ppr.value.transformby(ucs); list<double> radius = new list<double>(); radius.add(0.5); ...
Read more

ios - UITEXTFIELD InputView Uipicker not working in swift -

-
hello trying pop uipickerview view programmatically when user clicks on textfield. have tried doesn't doing anything. isn't working. nothing happening when click textfield class userprofiletableviewcontroller: uitableviewcontroller,uitextfielddelegate,uipickerviewdelegate,uipickerviewdatasource { var itempicker: uipickerview! = uipickerview() @iboutlet weak var gendertxtfield: uitextfield! var gender = ["male","female"] override func viewdidload() { super.viewdidload() gendertxtfield.delegate = self itempicker!.delegate = self itempicker!.datasource = self itempicker!.backgroundcolor = uicolor.blackcolor() self.gendertxtfield.inputview = itempicker } func numberofcomponentsinpickerview(pickerview: uipickerview) -> int{ return 1 } // returns # of rows in each component.. func pickerview(pickerview: uipickerview, numberofrowsincomponent component:...
Read more