Styjun

How can the US president give an order to a civilian?

Why is Skinner so awkward in Hot Fuzz?

Is there a risk to write an invitation letter for a stranger to obtain a Czech (Schengen) visa?

How to write a nice frame challenge?

How "fast" does astronomical events happen?

Are their examples of rowers who also fought?

How can I detect if I'm in a subshell?

How to know whether to write accidentals as sharps or flats?

Is there any effect in D&D 5e that cannot be undone?

How could I create a situation in which a PC has to make a saving throw or be forced to pet a dog?

Leveraging cash for buying car

Explicit direct #include vs. Non-contractual transitive #include

How can I maintain game balance while allowing my player to craft genuinely useful items?

My student in one course asks for paid tutoring in another course. Appropriate?

How Linux command "mount -a" works

How can a flywheel makes engine runs smoothly?

How to make a villain when your PCs are villains?

Right indicator flash-frequency has increased and rear-right bulb is out

How to ask if I can mow my neighbor's lawn

How do credit card companies know what type of business I'm paying for?

Testing thermite for chemical properties

How to prevent cables getting intertwined

How to avoid offending original culture when making conculture inspired from original

Why can't I craft scaffolding in Minecraft 1.14?

How to use specific ElasticSearch query according to SparkStreaming-Kafka message

Elasticsearch query to return all recordselasticsearch bool query combine must with ORSparkStreaming keep processing even no data in kafkaIs it possible to obtain specific message offset in Kafka+SparkStreaming?SparkStreaming/Kafka Offset handlingSparkStreaming put a kafka message into Hbasedo not want string as type when using foreach in scala spark streaming?[SparkStreaming]Kafka ConsumerRecord is not serializableSparkStreaming & Kafka: value reduceByKey is not a member of org.apache.spark.streaming.dstream.DStream[Any]Splitting Kafka Message Line by line in Spark Structured Streaming

.everyoneloves__top-leaderboard:empty,.everyoneloves__mid-leaderboard:empty,.everyoneloves__bot-mid-leaderboard:empty height:90px;width:728px;box-sizing:border-box;

0

I'm using SparkStreaming-Kafka, and want to support ElasticSearch real_time search with specific query, which is from Kafka message.

Code like bellow:

def creatingFuncTest():StreamingContext=

 val ssc = new StreamingContext(sc, Seconds(duration.toInt))
 ssc.checkpoint(checkpointDir)
 val kafkaParams = KafkaUtil.getKafkaParam(brokers, appName)

 val topics = actionTopicList.split(",").toSet

 val foodMessages = KafkaUtils
 .createDirectStream[String, String, StringDecoder, StringDecoder](
 ssc,
 kafkaParams,
 topics
 )


 val foodBatch: DStream[(String,Float, Float)] =
 foodMessages
 .filter(_._2.nonEmpty)
 .map msg =>
 try 
 println("___________ msg :" + msg._2)
 val gson = new Gson()
 val vo = gson.fromJson(msg._2, classOf[PoiMsg])
 (vo.person_id.toString, vo.latitude.toFloat,vo.longitude.toFloat)
 catch 
 case e: Exception =>
 println("____________" + e.getMessage)
 ("", 0.0f,0.0f)
 
 
 .filter(_._1.nonEmpty)



 foodBatch.foreachRDD(row =>
 row.foreach(t =>
 var lat = t._2
 var lon = t._3


 val query:String =s"""
 "filter" : 
 "geo_distance" : //...
 "distance" : "200km",
 "pin.location" : "lat" : "$lat", "lon" : "#lon" 
 
 """

 val rdd = sc.esRDD("recommend_diet_menu/fooddocument", query)

 println(rdd.count())

 )
 )
 ssc

I know in RDD, it's wrong to generate new RDD, but what's the right way?

apache-spark elasticsearch spark-streaming spark-streaming-kafka

share|improve this question

edited Mar 25 at 6:39

pirho

5,299122132

asked Mar 25 at 3:59

laclac

3016

add a comment | 

0

I'm using SparkStreaming-Kafka, and want to support ElasticSearch real_time search with specific query, which is from Kafka message.

Code like bellow:

def creatingFuncTest():StreamingContext=

 val ssc = new StreamingContext(sc, Seconds(duration.toInt))
 ssc.checkpoint(checkpointDir)
 val kafkaParams = KafkaUtil.getKafkaParam(brokers, appName)

 val topics = actionTopicList.split(",").toSet

 val foodMessages = KafkaUtils
 .createDirectStream[String, String, StringDecoder, StringDecoder](
 ssc,
 kafkaParams,
 topics
 )


 val foodBatch: DStream[(String,Float, Float)] =
 foodMessages
 .filter(_._2.nonEmpty)
 .map msg =>
 try 
 println("___________ msg :" + msg._2)
 val gson = new Gson()
 val vo = gson.fromJson(msg._2, classOf[PoiMsg])
 (vo.person_id.toString, vo.latitude.toFloat,vo.longitude.toFloat)
 catch 
 case e: Exception =>
 println("____________" + e.getMessage)
 ("", 0.0f,0.0f)
 
 
 .filter(_._1.nonEmpty)



 foodBatch.foreachRDD(row =>
 row.foreach(t =>
 var lat = t._2
 var lon = t._3


 val query:String =s"""
 "filter" : 
 "geo_distance" : //...
 "distance" : "200km",
 "pin.location" : "lat" : "$lat", "lon" : "#lon" 
 
 """

 val rdd = sc.esRDD("recommend_diet_menu/fooddocument", query)

 println(rdd.count())

 )
 )
 ssc

I know in RDD, it's wrong to generate new RDD, but what's the right way?

apache-spark elasticsearch spark-streaming spark-streaming-kafka

share|improve this question

edited Mar 25 at 6:39

pirho

5,299122132

asked Mar 25 at 3:59

laclac

3016

add a comment | 

I'm using SparkStreaming-Kafka, and want to support ElasticSearch real_time search with specific query, which is from Kafka message.

Code like bellow:

def creatingFuncTest():StreamingContext=

 val ssc = new StreamingContext(sc, Seconds(duration.toInt))
 ssc.checkpoint(checkpointDir)
 val kafkaParams = KafkaUtil.getKafkaParam(brokers, appName)

 val topics = actionTopicList.split(",").toSet

 val foodMessages = KafkaUtils
 .createDirectStream[String, String, StringDecoder, StringDecoder](
 ssc,
 kafkaParams,
 topics
 )


 val foodBatch: DStream[(String,Float, Float)] =
 foodMessages
 .filter(_._2.nonEmpty)
 .map msg =>
 try 
 println("___________ msg :" + msg._2)
 val gson = new Gson()
 val vo = gson.fromJson(msg._2, classOf[PoiMsg])
 (vo.person_id.toString, vo.latitude.toFloat,vo.longitude.toFloat)
 catch 
 case e: Exception =>
 println("____________" + e.getMessage)
 ("", 0.0f,0.0f)
 
 
 .filter(_._1.nonEmpty)



 foodBatch.foreachRDD(row =>
 row.foreach(t =>
 var lat = t._2
 var lon = t._3


 val query:String =s"""
 "filter" : 
 "geo_distance" : //...
 "distance" : "200km",
 "pin.location" : "lat" : "$lat", "lon" : "#lon" 
 
 """

 val rdd = sc.esRDD("recommend_diet_menu/fooddocument", query)

 println(rdd.count())

 )
 )
 ssc

I know in RDD, it's wrong to generate new RDD, but what's the right way?

apache-spark elasticsearch spark-streaming spark-streaming-kafka

share|improve this question

edited Mar 25 at 6:39

pirho

5,299122132

asked Mar 25 at 3:59

laclac

3016

def creatingFuncTest():StreamingContext=

 val ssc = new StreamingContext(sc, Seconds(duration.toInt))
 ssc.checkpoint(checkpointDir)
 val kafkaParams = KafkaUtil.getKafkaParam(brokers, appName)

 val topics = actionTopicList.split(",").toSet

 val foodMessages = KafkaUtils
 .createDirectStream[String, String, StringDecoder, StringDecoder](
 ssc,
 kafkaParams,
 topics
 )


 val foodBatch: DStream[(String,Float, Float)] =
 foodMessages
 .filter(_._2.nonEmpty)
 .map msg =>
 try 
 println("___________ msg :" + msg._2)
 val gson = new Gson()
 val vo = gson.fromJson(msg._2, classOf[PoiMsg])
 (vo.person_id.toString, vo.latitude.toFloat,vo.longitude.toFloat)
 catch 
 case e: Exception =>
 println("____________" + e.getMessage)
 ("", 0.0f,0.0f)
 
 
 .filter(_._1.nonEmpty)



 foodBatch.foreachRDD(row =>
 row.foreach(t =>
 var lat = t._2
 var lon = t._3


 val query:String =s"""
 "filter" : 
 "geo_distance" : //...
 "distance" : "200km",
 "pin.location" : "lat" : "$lat", "lon" : "#lon" 
 
 """

 val rdd = sc.esRDD("recommend_diet_menu/fooddocument", query)

 println(rdd.count())

 )
 )
 ssc

apache-spark elasticsearch spark-streaming spark-streaming-kafka

apache-spark elasticsearch spark-streaming spark-streaming-kafka

share|improve this question

edited Mar 25 at 6:39

pirho

5,299122132

asked Mar 25 at 3:59

laclac

3016

share|improve this question

edited Mar 25 at 6:39

pirho

5,299122132

asked Mar 25 at 3:59

laclac

3016

edited Mar 25 at 6:39

pirho

5,299122132

edited Mar 25 at 6:39

pirho

5,299122132

asked Mar 25 at 3:59

laclac

3016

asked Mar 25 at 3:59

laclac

3016