Getting RabbitMQ Message Delivery Tag

[mantovani]

https://stackoverflow.com/questions/47026864/getting-rabbitmq-message-delivery-tag-stratio-spark-rabbitmq

[kaveh-hariri]

Have you tried the below?

https://github.com/Stratio/spark-rabbitmq/blob/master/src/test/scala/org/apache/spark/streaming/rabbitmq/RabbitMQDistributedConsumerIT.scala

[mantovani]

Yes, I saw this on the test folder. And I still didn't see how I get the delivery tag on the stream.

…

On Tue, Oct 31, 2017 at 2:42 PM, kaveh-hariri ***@***.***> wrote: Have you tried the below? https://github.com/Stratio/spark-rabbitmq/blob/master/ src/test/scala/org/apache/spark/streaming/rabbitmq/ RabbitMQDistributedConsumerIT.scala — You are receiving this because you authored the thread. Reply to this email directly, view it on GitHub <#118 (comment)>, or mute the thread <https://github.com/notifications/unsubscribe-auth/AAC5fYmBfVY_UIA3B1cspKXLdeL5bobTks5sx2n4gaJpZM4QMud4> .

--

-- Daniel de Oliveira Mantovani Perl Evangelist/Data Hacker +1 786 459 1341

[kaveh-hariri]

I haven't tested this out, but it would be s omething like this --except change "body" with "deliveryTag"

val ssc = new StreamingContext(spark.sparkContext, Seconds(window))

import scala.collection.JavaConverters._
val stream = RabbitMQUtils.createDistributedStream[Map[String,Any]](ssc,distributedKey,rabbitMQParams,
  (rawMessage: Delivery) =>
    Map(
      "body" -> new Predef.String(rawMessage.getBody),
      "exchange" -> rawMessage.getEnvelope.getExchange,
      "routingKey" -> rawMessage.getEnvelope.getRoutingKey,
      "deliveryTag" -> rawMessage.getEnvelope.getDeliveryTag
    ) ++ {
      //Avoid null pointer Exception
      Option(rawMessage.getProperties.getHeaders) match {
        case Some(headers) => Map("headers" -> headers.asScala)
        case None => Map.empty[String, Any]
      }
    })
stream.start()
  stream.foreachRDD(rdd => { 
  new AvroIngest[v001.asset]().transformAvro(rdd.map(_.get("body").toString),spark,classOf[v001.asset].getName,dp_source,dp_source_type,avroKlass,outPath,Some(List("assetMetadata.data" -> "contextualMetadata")))
  })

ssc

---

edit: basically,
stream.foreachRDD(rdd => {rdd.map(dt => dosomething(dt.get("deliveryTag")))})

[mantovani]

Worked, thank you very much!

…

On Tue, Oct 31, 2017 at 3:42 PM, kaveh-hariri ***@***.***> wrote: I haven't tested this out, but it would be s omething like this --except change "body" with "deliveryTag" val ssc = new StreamingContext(spark.sparkContext, Seconds(window)) import scala.collection.JavaConverters._ val stream = RabbitMQUtils.createDistributedStream[Map[String,Any]](ssc,distributedKey,rabbitMQParams, (rawMessage: Delivery) => Map( "body" -> new Predef.String(rawMessage.getBody), "exchange" -> rawMessage.getEnvelope.getExchange, "routingKey" -> rawMessage.getEnvelope.getRoutingKey, "deliveryTag" -> rawMessage.getEnvelope.getDeliveryTag ) ++ { //Avoid null pointer Exception Option(rawMessage.getProperties.getHeaders) match { case Some(headers) => Map("headers" -> headers.asScala) case None => Map.empty[String, Any] } }) stream.start() stream.foreachRDD(rdd => { new AvroIngest[v001.asset]().transformAvro(rdd.map(_.get("body").toString),spark,classOf[v001.asset].getName,dp_source,dp_source_type,avroKlass,outPath,Some(List("assetMetadata.data" -> "contextualMetadata"))) }) ssc — You are receiving this because you authored the thread. Reply to this email directly, view it on GitHub <#118 (comment)>, or mute the thread <https://github.com/notifications/unsubscribe-auth/AAC5fShyWsVQoSGxJLxLU4oN-_zvNBI_ks5sx3gsgaJpZM4QMud4> .

--

-- Daniel de Oliveira Mantovani Perl Evangelist/Data Hacker +1 786 459 1341