如何将mqtt payload数据转换为kafka字符串类型

Question

现在我正在尝试从 mqtt 消息连接到 kafka（实际上是 kafka 中的 spark streaming）

我使用了这个连接器https://github.com/evokly/kafka-connect-mqtt

和 Spark-2.1.0，Kafka - 0.10.1.1

像这样的 Spark 流输出

({"schema":{"type":"string","optional":false},"payload":"mqtt"},{"schema":{"type":"bytes","optional":false},"payload":"MTIzMTIz"})

和生产者代码

object mqttProducer {
def main(args: Array[String]) {
val brokerUrl = "tcp://ip"
val topic = "mqtt"
val msg = "123123"

var client: MqttClient = null

// Creating new persistence for mqtt client
val persistence = new MqttDefaultFilePersistence("/tmp")

try {
  // mqtt client with specific url and client id
  client = new MqttClient(brokerUrl, MqttClient.generateClientId, persistence)

  client.connect()

  val msgTopic = client.getTopic(topic)
  val message = new MqttMessage(msg.getBytes("utf-8"))

  while (true) {
    msgTopic.publish(message)
    println("Publishing Data, Topic : %s, Message : %s".format(msgTopic.getName, message))
    Thread.sleep(1000)
  }
}

catch {
  case e: MqttException => println("Exception Caught: " + e)
}

finally {
  client.disconnect()
}

和 spark-streaming kafka 消费者代码

package hb.test1
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.SparkConf
import org.apache.spark.streaming.Seconds
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
import org.apache.spark.streaming.kafka010.KafkaUtils
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent

object test2 {

  def main(args: Array[String]): Unit = {

val sparkConf = new SparkConf().setAppName("app")
val ssc = new StreamingContext(sparkConf, Seconds(1))    


val kafkaParams = Map[String, Object](
  "bootstrap.servers" -> 
  "servers ip",
  "key.deserializer" -> classOf[StringDeserializer],
  "value.deserializer" -> classOf[StringDeserializer],
  "group.id" -> "use_a_separate_group_id_for_each_stream",
  "auto.offset.reset" -> "latest",
  "enable.auto.commit" -> (false: java.lang.Boolean)
)
val topics = Array("mqtt-kafka")
  val stream = KafkaUtils.createDirectStream[String, String](
  ssc,
  PreferConsistent,
  Subscribe[String, String](topics, kafkaParams)
)

val testStream = stream.map(x => (x.key, x.value))


testStream.print()

ssc.start()
ssc.awaitTermination()
  }
}

我怎么能得到字符串而不是字节？请大家帮忙

Answer 1

有效载荷 "MTIzMTIz" 是字符串“123123”，只是经过 base64 编码。如果你只想获取 MQTT 负载并将其发送到 Kafka 而无需 base64 编码，你应该使用 ByteArrayConverter。在我对同一个 mqtt 连接器的配置中，我像这样设置值转换器：

"value.converter": "io.confluent.connect.replicator.util.ByteArrayConverter"

上述 ByteArrayConverter 随 Confluent Enterprise 发行版一起提供，但还有其他开源 Kafka Connect ByteArrayConverter，例如 qubole/streamx kafka-connect-s3 连接器中包含的那个。

https://github.com/qubole/streamx/blob/8b9d43008d9901b8e6020404f137944ed97522e2/src/main/java/com/qubole/streamx/ByteArrayConverter.java

有 KIP-128 可以将标准的 ByteArrayConverter 添加到 Kafka Connect 框架

https://cwiki.apache.org/confluence/display/KAFKA/KIP-128%3A+Add+ByteArrayConverter+for+Kafka+Connect

更新：Kafka 0.11 现已发布并附带了一个 ByteArrayConverter。配置 "value.converter": "org.apache.kafka.connect.converters.ByteArrayConverter"，你应该得到没有 Base64 编码的原始 mqtt 负载。

如何将mqtt payload数据转换为kafka字符串类型

How to conver mqtt's payload data to kafka string type

mqtt

apache-kafka

apache-kafka-connect