scala - How to fetch Kafka Stream and print it in Spark Shell? -


first built sbt in folder in following way

val sparkversion = "1.6.3"                                                                                                                            scalaversion := "2.10.5"                                                                                                                              resolvers += "spark packages repo" @ "https://dl.bintray.com/spark-packages/maven"                                                                   librarydependencies ++= seq(                                                                                                                          "org.apache.spark" %% "spark-streaming" % sparkversion,                                                                                               "org.apache.spark" %% "spark-streaming-kafka" % sparkversion                                                                                           )                                                                                                                                                    librarydependencies +="datastax" % "spark-cassandra-connector" % "1.6.3-s_2.10"                                                                       librarydependencies +="org.apache.spark" %% "spark-sql" % "1.1.0"

later in same folder "build.sbt" exists started spark shell in following way 

 >/usr/hdp/2.6.0.3-8/spark/bin/spark-shell --packages datastax:spark-cassandra-connector:1.6.3-s_2.10 --conf spark.cassandra.connection.host=127.0.0.1

these warnings shown while spark shell started:

warn abstractlifecycle: failed selectchannelconnector@0.0.0.0:4040: java.net.bind java.net.bindexception: address in use warn abstractlifecycle: failed org.spark-project.jetty.server.server@75bf9e67: java.net.bindexception: address in use

in spark shell importing following packages 

import org.apache.spark.sparkconf; import org.apache.spark.streaming.streamingcontext; import org.apache.spark.streaming.seconds; import org.apache.spark.streaming.kafka.kafkautils; import com.datastax.spark.connector._ ; import org.apache.spark.sql.cassandra._ ;

then in spark shell creating configuration in below way:

val conf = new sparkconf().setmaster("local[*]").setappname("kafkareceiver").set("spark.driver.allowmultiplecontexts", "true").setmaster("local");

after creating configuration assigning it, created new spark streaming context in below way:

val ssc = new streamingcontext(conf, seconds(10))

during creation of spark streaming context few warnings shown above raised again along other warning, shown below

warn abstractlifecycle: failed selectchannelconnector@0.0.0.0:4040: java.net.bind java.net.bindexception: address in use . . . warn abstractlifecycle: failed org.spark-project.jetty.server.server@75bf9e67: java.net.bindexception: address in use . . . warn sparkcontext: multiple running sparkcontexts detected in same jvm!                                                         org.apache.spark.sparkexception: 1 sparkcontext may running in jvm (see spark-2243). ignore error, set spark.driver.allowmulti plecontexts = true. running sparkcontext created at:                                                                                org.apache.spark.sparkcontext.<init>(sparkcontext.scala:82)                                                                                           org.apache.spark.repl.sparkiloop.createsparkcontext(sparkiloop.scala:1017) . . . warn streamingcontext: spark.master should set local[n], n > 1 in local mode if have receivers data, otherwise spa rk jobs not resources process received data.                                                                                          ssc: org.apache.spark.streaming.streamingcontext = org.apache.spark.streaming.streamingcontext@616f1c2e

then using created spark streaming context created kafkastream in below way

val kafkastream = kafkautils.createstream(ssc, "localhost:2181","spark-streaming-consumer-group", map("spark-topic" -> 5))

then printing stream , starting ssc in below way 

kafkastream.print() ssc.start

after use of above command in shell output shown in below images

image1- eariler part of output

image2 - after fetching jars & connecting zookeeper

issues came across in zookeeper

later starts following mess ! of stream without printing values information shown in below image final output

output that's getting printed repeatedly here shown below !

17/08/18 10:01:30 info jobscheduler: starting job streaming job 1503050490000 ms.0 job set of time 1503050490000 ms                              17/08/18 10:01:30 info jobscheduler: finished job streaming job 1503050490000 ms.0 job set of time 1503050490000 ms                              17/08/18 10:01:30 info jobscheduler: total delay: 0.003 s time 1503050490000 ms (execution: 0.000 s)                                              17/08/18 10:01:30 info blockrdd: removing rdd 3 persistence list                                                                                 17/08/18 10:01:30 info kafkainputdstream: removing blocks of rdd blockrdd[3] @ createstream @ <console>:39 of time 1503050490000 ms                 17/08/18 10:01:30 info receivedblocktracker: deleting batches arraybuffer(1503050470000 ms)                                                           17/08/18 10:01:30 info inputinfotracker: remove old batch metadata: 1503050470000 ms                                                                  17/08/18 10:01:30 info blockmanager: removing rdd 3                                                                                                   17/08/18 10:01:40 info jobscheduler: added jobs time 1503050500000 ms                                                                             -------------------------------------------                                                                                                           time: 1503050500000 ms                                                                                                                                ------------------------------------------- 

warn abstractlifecycle: failed selectchannelconnector@0.0.0.0:4040: java.net.bind    java.net.bindexception: address in use

it means port needs in use. rule, port 4040 used spark-thrifteserver. try stop thriftserver using stop-thriftserver.sh spark/sbin folder. or check else use port , free it.


Comments

Post a Comment

Popular posts from this blog

Is there a better way to structure post methods in Class Based Views -

i'm using class based views in django , never specify url in ajax calls , omit action parameter in form because know these requests go through post method of class based view. notice view - profile page make many requests depending on user updating on page - post method become convoluted , if/else mess example: def post(self, request, *args, **kwargs): # handle user changing available date if request.post.get('availabledate') != none: self.updateavailabledate(request) return jsonresponse({'result':'success'}) if request.post.get('newprojectname') != none: creative_user = creativeuserprofile.objects.get(id = request.user.id) project = project.create_project(creative_user, request.post['newprojectname']) project.save() return jsonresponse({'projectid': project.id }) if request.post.get('projectimage') != none , request.files['file']: proj...
Read more

performance - Why is XCHG reg, reg a 3 micro-op instruction on modern Intel architectures? -

i'm doing micro-optimization on performance critical part of code , came across sequence of instructions (in at&t syntax): add %rax, %rbx mov %rdx, %rax mov %rbx, %rdx i thought had use case xchg allow me shave instruction , write: add %rbx, %rax xchg %rax, %rdx however, dimay found agner fog's instruction tables , xchg 3 micro-op instruction 2 cycle latency on sandy bridge, ivy bridge, broadwell, haswell , skylake. 3 whole micro-ops , 2 cycles of latency! 3 micro-ops throws off 4-1-1-1 cadence , 2 cycle latency makes worse original in best case since last 2 instructions in original might execute in parallel. now... cpu might breaking instruction micro-ops equivalent to: mov %rax, %tmp mov %rdx, %rax mov %tmp, %rdx where tmp anonymous internal register , suppose last 2 micro-ops run in parallel latency 2 cycles. given register renaming occurs on these micro-architectures, though, doesn't make sense me done way. why wouldn't register renamer ...
Read more

c# - Asp.net web api : redirect unauthorized requst to forbidden page -

im trying redirect unauthorized request forbidden page instead i'm getting forbidden page in response body , how can fix ? here's startup class : app.createperowincontext(storecontext.create); app.createperowincontext<applicationusermanager>(applicationusermanager.create); app.createperowincontext<applicationsigninmanager>(applicationsigninmanager.create); app.usecookieauthentication(new cookieauthenticationoptions { authenticationtype = defaultauthenticationtypes.applicationcookie, expiretimespan = timespan.fromdays(30), }); app.useexternalsignincookie(defaultauthenticationtypes.externalcookie); app.usetwofactorsignincookie(defaultauthenticationtypes.twofactorcookie, timespan.fromminutes(5)); app.usetwofactorrememberbrowsercookie(defaultauthenticationtypes.twofactorrememberbrowsercookie); the method im trying reach : [httpget] [authorize(roles = "admin")] public string getcurrentusername() { return userman...
Read more