#scala #spark-streaming #mqtt
#scala #потоковая передача spark #mqtt
Вопрос:
У меня есть требование провести анализ с использованием spark данных, поступающих с устройства Интернета вещей через брокер MQTT. Подключение из моей работы spark осуществляется к брокеру MQTT, где я могу подписаться на определенную тему. Я использовал библиотеку MQTTUtils в spark для подключения к брокеру, но у меня есть сомнения относительно того, как библиотека работает внутри. Я заметил, что «MQTTutils.createStream» подключается к брокеру MQTT для темы. В этом случае, если мне нужно подписаться на 100 тем в брокере MQTT, он может установить 100 подключений к брокеру. Это нежелательно в реальном сценарии. Пожалуйста, дайте мне знать, если это работает не так, как я думаю.
Итак, я решил создать пользовательский приемник для брокера MQTT, чтобы я мог управлять подключением в моем клиенте MQTT. Я просмотрел документ о том, как реализовать пользовательский приемник, но мне не удалось реализовать его надлежащим образом.
Если у кого-то есть опыт работы с таким пользовательским приемником, пожалуйста, помогите мне заставить его работать.
Ценю вашу поддержку, поскольку она имеет решающее значение для моего решения
Комментарии:
1. Этот вопрос слишком расплывчатый, вам нужно объяснить, что вы на самом деле сделали и как это не сработало (показать код / ошибки), и кто-нибудь может помочь вам это исправить. Просто попросить совета или попросить кого-нибудь предоставить рабочий код не по теме.
2. В настоящее время мой код хорошо работает с библиотекой MQTTUtils, но я столкнулся с ее ограничением. «MQTTutils.createStream» каждый раз подключается к MQTTbroker для темы, и в нем нет параметров аутентификации. Итак, я думаю о пользовательском приемнике для потоковой передачи данных от клиента MQTT таким образом, чтобы он устанавливал только 1 соединение с брокером с аутентификацией и мог подписываться на различные темы и транслировать все подписанные темы в контекст потоковой передачи spark.