apache spark - How to get size (metadata) of a message without fetch request in Kafka through SimpleConsumer? -

-

i using simpleconsumer , trying pull message size wise (bytes) using spark.

i able earliest , latest offset using metadata request, dont know how number of bytes in kafka (0.8.0).

i dont want use fetch request because want check sufficient data run spark job (not streaming) perform action.

from metadata response cant size, size need download message meaningless.


Comments

Post a Comment

Popular posts from this blog

android - questions about switching from C2DM to GCM -

-
i'm converting app c2dm gcm , have questions. i've modified gcmquickstart example found here: git clone https://github.com/googlesamples/google-services.git in class extends application, i've got: public class myapp extends application { @override public void oncreate() { super.oncreate(); if (this.checkplayservices()) { intent intent = new intent(this, registrationintentservice.class); startservice(intent); } else{ log.e(tag,"failed checkforgcm"); } } } afaik, starts service , it'll run forever or if operating system shuts down whereupon operating system restart it. correct? checkplayservices() makes sure android 2.3 or greater , google play services installed these requirements use gcm. correct? public class registrationintentservice extends intentservice{ private static final string tag = "regintentservice"; public registratio...
Read more

c++ - rosrun via ssh doesnt start a node properly -

-
we working ros , try run node via ssh in shell script. have controlling (master) computer on execute script following line: ssh user@xxx.xxx.x.x -x "/bin/bash -c 'source /etc/profile; rosrun stargazer_alter stargazer_node'" the node getting started (it occurs in "rosnode list" , node listed publisher in "rostopic info") main method seems not executed (neither print lines show nor messages published). #include "stargazer_listener.h" enum states {config, readdata}; int main(int argc, char** argv) { ros_info("test!!!! stargazer starts!"); ros::init(argc, argv, "stargazer_node"); stargazer_listener starg; int ifd = 0; char *rec_msg = (char*)malloc(buffsize * sizeof(char)); ifd = starg.portsetup(); states state = config; while(ros::ok()) { switch(state) { case config: starg.cmd(ifd, rec_msg, "~#calcstop`"); printf("calcstop\n"); s...
Read more

How to resolve Delphi error: Incompatible types: 'PWideChar' and 'Pointer' -

-
i have piece of code delphi 7: var lprgndata: prgndata; pc: pchar; pr: prect; ... pc := @(lprgndata^.buffer[0]); in delphi xe4 gives following compile error: incompatible types: 'pwidechar' , 'pointer' how should code updated work correctly in xe4? thanks whether or not compiles depends upon setting of type-checked pointers option . have enabled option excellent decision. doing results in stricter type checking. with type-checked pointers disabled, code compile. type-checked pointers enabled, code not compile, want because code not valid. now, on types in question. defined in windows unit this: type prgndata = ^trgndata; {$externalsym _rgndata} _rgndata = record rdh: trgndataheader; buffer: array[0..0] of byte; reserved: array[0..2] of byte; end; trgndata = _rgndata; {$externalsym rgndata} rgndata = _rgndata; the benefit of using type-checked pointers compiler can tell doing not valid. knows lprgndat...
Read more