Расшифровать сообщение Kafka-mqtt - PullRequest
Новая
       8

Расшифровать сообщение Kafka-mqtt

0 голосов
/ 28 августа 2018

Я получаю сообщения от брокера mqtt по теме kafka, используя соединитель kafka-mqtt. Затем я прочитал это сообщение из темы Кафки, используя пользователя Кафки на спарк. Когда я печатаю сообщения, это результат. Как я могу правильно прочитать сообщения?

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

SparkConf sparkConf = new SparkConf().setAppName("GestoreSoccorso").setMaster("local[2]");
JavaStreamingContext ssc = new JavaStreamingContext(sparkConf, new Duration(500));

Map<String, Object> kafkaParams = new HashMap<>();
kafkaParams.put("bootstrap.servers", "10.0.4.215:9092");
kafkaParams.put("key.deserializer", StringDeserializer.class);
kafkaParams.put("value.deserializer", StringDeserializer.class);
kafkaParams.put("group.id", "use_a_separate_group_id_for_each_stream");
kafkaParams.put("auto.offset.reset", "earliest");
kafkaParams.put("enable.auto.commit", false);
Collection<String> topics = Arrays.asList("ankioverdrive_v1_events");

JavaInputDStream<ConsumerRecord<String, String>> stream =
        KafkaUtils.createDirectStream(
                ssc,
                LocationStrategies.PreferConsistent(),
                ConsumerStrategies.<String, String>Subscribe(topics, kafkaParams)
        );

Тогда этот код я использую, чтобы читать сообщения из темы и распечатывать их. 

stream.foreachRDD(new VoidFunction<JavaRDD<ConsumerRecord<String, String>>>() {
            @Override
            public void call(JavaRDD<ConsumerRecord<String, String>> consumerRecordJavaRDD) throws Exception {
                consumerRecordJavaRDD.foreach(new VoidFunction<ConsumerRecord<String, String>>() {
                    @Override
                    public void call(ConsumerRecord<String, String> stringStringConsumerRecord) throws Exception {
                        String stringa=stringStringConsumerRecord.value();
                        System.out.println(DEBUG+"DATI RICEVUTI -> "+ stringa);

Наконец, это вывод 

DEBUG: DATI RICEVUTI ->     ܑ�՛Y
SKULL   �    unknown0%

1 Ответ

0 голосов
/ 28 августа 2018

Вы передаете неправильные значения для параметров key.deserializer и value.deserializer.

Вместо 

kafkaParams.put("key.deserializer", StringDeserializer.class);
kafkaParams.put("value.deserializer", StringDeserializer.class);

нужно пройти 

kafkaParams.put("key.deserializer", StringDeserializer.class.getName());
kafkaParams.put("value.deserializer", StringDeserializer.class.getName());

или просто 

kafkaParams.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
kafkaParams.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
...