我使用kafka mqtt连接器从一个kafka主题的mqtt代理接收消息。然后我使用spark上的kafka消费者阅读了kafka主题中的这些信息。当我打印消息时,这就是结果。我怎样才能正确地阅读这些信息?
这是set consumer和create stream的代码。
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条答案
按热度按时间vmdwslir1#
您为参数传递了错误的值
key.deserializer
以及value.deserializer
.而不是
你需要通过
或者只是