apache spark streaming - kafka - reading older messages -

-

i trying read older messages kafka spark streaming. however, able retrieve messages sent in real time (i.e., if populate new messages, while spark program running - messages).

i changing groupid , consumerid make sure zookeeper isn't not giving messages knows program has seen before.

assuming spark seeing offset in zookeeper -1, shouldn't read old messages in queue? misunderstanding way kafka queue can used? i'm new spark , kafka, can't rule out i'm misunderstanding something. 

package com.kibblesandbits  import org.apache.spark.sparkcontext import org.apache.spark.streaming.{seconds, streamingcontext} import org.apache.spark.streaming.kafka.kafkautils  import net.liftweb.json._  object kafkastreamingtest {    val cfg = new configloader().load   val zookeeperhost = cfg.zookeeper.host   val zookeeperport = cfg.zookeeper.port   val zookeeper_kafka_chroot = cfg.zookeeper.kafka_chroot    implicit val formats = defaultformats     def parser(json: string): string = {     return json }  def main(args : array[string]) {   val zkquorum = "test-spark02:9092"    val group = "mygroup99"   val topic = map("testtopic" -> 1)   val sparkcontext = new sparkcontext("local[3]", "kafkaconsumer1_new")   val ssc = new streamingcontext(sparkcontext, seconds(3))   val json_stream = kafkautils.createstream(ssc, zkquorum, group, topic)   var gp = json_stream.map(_._2).map(parser)    gp.saveastextfiles("/tmp/sparkstreaming/mytest", "json")   ssc.start() }

when running this, see following message. confident it's not not seeing messages because offset set.

14/12/05 13:34:08 info consumerfetchermanager: [consumerfetchermanager-1417808045047] added fetcher partitions arraybuffer([[testtopic,0], initoffset -1 broker id:1,host:test-spark02.vpc,port:9092] , [[testtopic,1], initoffset -1 broker d:1,host:test-spark02.vpc,port:9092] , [[testtopic,2], initoffset -1 broker id:1,host:test-spark02.vpc,port:9092] , [[testtopic,3], initoffset -1 broker id:1,host:test-spark02.vpc,port:9092] , [[testtopic,4], initoffset -1 broker id:1,host:test-spark02.vpc,port:9092] )

then, if populate 1000 new messages -- can see 1000 messages saved in temp directory. don't know how read existing messages, should number in (at point) tens of thousands.

use alternative factory method on kafkautils lets provide configuration kafka consumer:

def createstream[k: classtag, v: classtag, u <: decoder[_]: classtag, t <: decoder[_]: classtag](       ssc: streamingcontext,       kafkaparams: map[string, string],       topics: map[string, int],       storagelevel: storagelevel     ): receiverinputdstream[(k, v)]

then build map kafka configuration , add parameter 'kafka.auto.offset.reset' set 'smallest':

val kafkaparams = map[string, string](       "zookeeper.connect" -> zkquorum, "group.id" -> groupid,       "zookeeper.connection.timeout.ms" -> "10000",       "kafka.auto.offset.reset" -> "smallest" )

provide config factory method above. "kafka.auto.offset.reset" -> "smallest" tells consumer starts smallest offset in topic.


Comments

Post a Comment

Popular posts from this blog

python - mat is not a numerical tuple : openCV error -

-
i have write down code showing error ma not getting it: please help: showing mat not numerical tuple: import cv import cv2 capture = cv2.videocapture("j.3gp") while(1): _, frame1 = capture.read() grayimage1 = cv2.cvtcolor(frame1, cv2.color_bgr2gray) _, frame2 = capture.read() grayimage2 = cv2.cvtcolor(frame2, cv2.color_bgr2gray) differenceimage = cv2.absdiff(grayimage1,grayimage2) thresholdimage = cv2.threshold(differenceimage,25,255,cv2.thresh_binary) cv2.imshow("difference image", differenceimage) cv2.imshow("threshold image", thresholdimage) cv2.imshow("image", frame1) k = cv2.waitkey(30) & 0xff error arising : ----------------------------------------------------------------------------------------- traceback (most recent call last): file "desk.py", line 15, in <module> cv2.imshow("threshold image", thresholdimage) typeerror: mat not numerical tuple ...
Read more

c# - MSAA finds controls UI Automation doesn't -

-
i'm working on automating windows application. i'm using teststack white framework. i've hit problem. program has 'window' object cannot see inside of. white shows no controls inside of it. inspect.exe shows no controls inside of either when running in ui automation mode. if switch inspect msaa see controls inside fine. there anyway me use msaa c# handle on these controls? if can identify msaa functions need, can use p/invoke call them c#. here's example article doing msaa: http://www.codeproject.com/articles/38906/ui-automation-using-microsoft-active-accessibility also, pinvoke.net can used identify iaccessible (msaa) functions: http://www.pinvoke.net/search.aspx?search=iaccessible&namespace=[all] here's answer along lines: msaa com-based? finally, alternative p/invoke, might able use tlbimp.exe create wrapper assembly oleacc.dll, , access msaa functions through it. i'm not sure if works msaa, it's worth try. as e...
Read more

wordpress - .htaccess: RewriteRule: bad flag delimiters -

-
pulling hair out on one. have following .htaccess file custom redirects in addition wordpress's default settings. when tested on local wamp server worked fine after moving production i'm not getting rewriterule: bad flag delimiters error in server log , after while site goes down internal server error any appreciated. in advance! <ifmodule mod_rewrite.c> rewriteengine on rewritebase / rewriterule ^rockford_weddings___welcome\.html$ /wedding/ [r=301,l] rewriterule ^blog/?$ /journal/ [r=301,l] rewriterule ^blog/(.*)$ /$1 [r=301,l] rewriterule ^portraitinvestment/?$ /portraitinvestment\.pdf [r=301,l] rewriterule ^weddinginvestment/?$ /weddinginvestment\.pdf [r=301,l] rewriterule ^holidaycards2014/?$ /holidaycards2014\.pdf [r=301,l] </ifmodule> # begin wordpress <ifmodule mod_rewrite.c> rewriteengine on rewritebase / rewriterule ^index\.php$ - [l] rewritecond %{request_filename} !-f rewritecond %{request_filename} !-d rewriterule . /index.php [l] </...
Read more