Great point :) Cui, Here’s a cleaner way than I had before, w/out the use of spark sql for the mapping:
KafkaUtils.createStream[String, String, StringDecoder, StringDecoder]( ssc, kafka.kafkaParams, Map("github" -> 5), StorageLevel.MEMORY_ONLY) .map{ case (k,v) => JsonParser.parse(v).extract[MonthlyCommits]} .saveToCassandra("githubstats","monthly_commits") HELENA EDELSON Senior Software Engineer, DSE Analytics On Mar 5, 2015, at 9:33 AM, Ted Yu <yuzhih...@gmail.com> wrote: > Cui: > You can check messages.partitions.size to determine whether messages is an > empty RDD. > > Cheers > > On Thu, Mar 5, 2015 at 12:52 AM, Akhil Das <ak...@sigmoidanalytics.com> wrote: > When you use KafkaUtils.createStream with StringDecoders, it will return > String objects inside your messages stream. To access the elements from the > json, you could do something like the following: > > > val mapStream = messages.map(x=> { > val mapper = new ObjectMapper() with ScalaObjectMapper > mapper.registerModule(DefaultScalaModule) > > mapper.readValue[Map[String,Any]](x).get("time") > }) > > > > Thanks > Best Regards > > On Thu, Mar 5, 2015 at 11:13 AM, Cui Lin <cui....@hds.com> wrote: > Friends, > > I'm trying to parse json formatted Kafka messages and then send back to > cassandra.I have two problems: > I got the exception below. How to check an empty RDD? > Exception in thread "main" java.lang.UnsupportedOperationException: empty > collection > at org.apache.spark.rdd.RDD$$anonfun$reduce$1.apply(RDD.scala:869) > at org.apache.spark.rdd.RDD$$anonfun$reduce$1.apply(RDD.scala:869) > at scala.Option.getOrElse(Option.scala:120) > at org.apache.spark.rdd.RDD.reduce(RDD.scala:869) > at org.apache.spark.sql.json.JsonRDD$.inferSchema(JsonRDD.scala:57) > at org.apache.spark.sql.SQLContext.jsonRDD(SQLContext.scala:232) > at org.apache.spark.sql.SQLContext.jsonRDD(SQLContext.scala:204) > > val messages = KafkaUtils.createStream[String, String, StringDecoder, > StringDecoder](…) > messages.foreachRDD { rdd => > val message:RDD[String] = rdd.map { y => y._2 } > sqlContext.jsonRDD(message).registerTempTable("tempTable") > sqlContext.sql("SELECT time,To FROM tempTable") > .saveToCassandra(cassandra_keyspace, cassandra_table, SomeColumns("key", > "msg")) > } > > 2. how to get all column names from json messages? I have hundreds of columns > in the json formatted message. > > Thanks for your help! > > > > > Best regards, > > Cui Lin > >