java - Java Kafka 消费者和 avro 反序列化器
问题描述
我正在开发一个带有火花流的简单 java。
我配置了一个 kafka jdbc 连接器(postgres 到主题),我想用火花流消费者阅读它。
我能够正确阅读主题:
./kafka-avro-console-consumer --bootstrap-server localhost:9092 --property schema.registry.url=http://localhost:8081 --property print.key=true --from-beginning --topic postgres-ip_audit
得到这个结果:
空 {"id":1557,"ip":{"string":"90.228.176.138"},"create_ts":{"long":1554819937582}}
当我将我的 java 应用程序与此配置一起使用时:
Map<String, Object> kafkaParams = new HashMap<>();
kafkaParams.put("bootstrap.servers", "localhost:9092");
kafkaParams.put("key.deserializer", StringDeserializer.class);
kafkaParams.put("value.deserializer", StringDeserializer.class);
kafkaParams.put("group.id", "groupStreamId");
kafkaParams.put("auto.offset.reset", "latest");
kafkaParams.put("enable.auto.commit", false);
我得到这样的结果:
�179.20.119.53�����Z
有人可以指出我如何解决我的问题吗?
我也尝试使用 ByteArrayDeserializer 并将 bytes[] 转换为字符串,但我总是得到错误的字符结果。
解决方案
您可以使用并拥有架构注册表来反序列化 avro 消息io.confluent.kafka.serializers.KafkaAvroDeserializer
以管理记录架构。
这是一个示例代码片段
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import io.confluent.kafka.serializers.KafkaAvroDecoder;
import kafka.serializer.StringDecoder;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.api.java.JavaPairInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import org.apache.spark.streaming.kafka.KafkaUtils;
import scala.Tuple2;
public class SparkStreaming {
public static void main(String... args) {
SparkConf conf = new SparkConf();
conf.setMaster("local[2]");
conf.setAppName("Spark Streaming Test Java");
JavaSparkContext sc = new JavaSparkContext(conf);
JavaStreamingContext ssc = new JavaStreamingContext(sc, Durations.seconds(10));
processStream(ssc, sc);
ssc.start();
ssc.awaitTermination();
}
private static void processStream(JavaStreamingContext ssc, JavaSparkContext sc) {
System.out.println("--> Processing stream");
Map<String, String> props = new HashMap<>();
props.put("bootstrap.servers", "localhost:9092");
props.put("schema.registry.url", "http://localhost:8081");
props.put("group.id", "spark");
props.put("specific.avro.reader", "true");
props.put("value.deserializer", "io.confluent.kafka.serializers.KafkaAvroDeserializer");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
Set<String> topicsSet = new HashSet<>(Collections.singletonList("test"));
JavaPairInputDStream<String, Object> stream = KafkaUtils.createDirectStream(ssc, String.class, Object.class,
StringDecoder.class, KafkaAvroDecoder.class, props, topicsSet);
stream.foreachRDD(rdd -> {
rdd.foreachPartition(iterator -> {
while (iterator.hasNext()) {
Tuple2<String, Object> next = iterator.next();
Model model = (Model) next._2();
System.out.println(next._1() + " --> " + model);
}
}
);
});
}
}
推荐阅读
- puppeteer - 如何将 Chrome 添加到容器以克服 circleCI 中的错误“无法启动 chrome”
- react-native - VS Code 实时分享反应原生?
- c++ - 为什么使用多重继承时我的 Hippomock 期望失败
- python - AttributeError:模块“tensorflow”没有属性“assign”
- tensorflow - LSTM 层输出大小与 KERAS 中的隐藏状态大小
- php - foreach 中未定义的属性,除非 print_r'd
- php - .htaccess FilesMatch 以阻止 php 执行名称包含大写字母的 php 文件
- cassandra - 增加 roles_validity_in_ms 和 permissions_validity_in_ms 的影响
- ios - 在 Xcode 11 中通过 Fastlane 构建 SPM 包不起作用
- c# - OpenXML 将行追加到 Word 中的现有表