Не удается создать поток MQTT с именем пользователя и паролем

Question

Я хочу создать поток MQTT, который использует имя пользователя и пароль для подписки. Я использую apache-bahir для создания потока mqtt. Когда я использовал метод MQTTUtils.createStream(), он принимает только ssc,brokerurl,topic и StorageLevel в качестве параметров. Когда я дал имя пользователя и пароль, он показывает "Не удается разрешить символ createStream ().

val lines = MQTTUtils.createStream(ssc, brokerUrl, topic, storageLevel, clientId, username, password, cleanSession, qos, connectionTimeout, keepAliveInterval, mqttVersion)

https://bahir.apache.org/docs/spark/2.3.0/spark-streaming-mqtt/

Эту страницу я рецензировал для реализации.

val sc = new SparkContext()
val ssc = new StreamingContext(sc,Seconds(10))
val stream = MQTTUtils.createStream(ssc,"broker.mqttdashboard.com","tag_topic",StorageLevel.DISK_ONLY,"clientid","username","password")

Сообщение об ошибке не удается разрешить символ MQTTUtils.createStream()

Answer 1

Попробуйте это:

 val spark  = SparkSession
      .builder()
      .appName("app")
      .config("spark.master", "local[*]")
      .getOrCreate()

val lines = spark.readStream
  .format("org.apache.bahir.sql.streaming.mqtt.MQTTStreamSourceProvider")
  .option("username", username)
  .option("password", password)
  .option("topic", topic)
  .option("brokerUrl",brokerUrl )
  .load()

у вас есть вся информация на Github

Answer 2

Используя JavaStreamingContext, мы можем решить эту проблему

val jssc = new JavaStreamingContext(sc,Seconds(2))
val clientId = MqttClient.generateClientId()
val stream =  MQTTUtils.createPairedByteArrayStream(jssc,brokerUrl,topic,clientId,username,password,true)
stream.print()
jssc.start()
jssc.awaitTermination()