我已经升级了我的Kafka和KafkaSpark流,但我面临着一些挑战,而改变他们的一些方法。像kafkautils一样,迭代器也在抛出错误。我的Kafka版本是0.10.1.1。所以,如果有人知道如何修复这些变化,那就太好了。谢谢
fykwrbwg1#
以前的kafkautils包是“org.apache.spark.streaming.kafka”。最新的包是“org.apache.spark.streaming.kafka010”。要设置kafkaparams和主题详细信息,请检查以下代码段,
import java.util.*;import org.apache.spark.SparkConf;import org.apache.spark.TaskContext;import org.apache.spark.api.java.*;import org.apache.spark.api.java.function.*;import org.apache.spark.streaming.api.java.*;import org.apache.spark.streaming.kafka010.*;import org.apache.kafka.clients.consumer.ConsumerRecord;import org.apache.kafka.common.TopicPartition;import org.apache.kafka.common.serialization.StringDeserializer;import scala.Tuple2;Map<String, Object> kafkaParams = new HashMap<>();kafkaParams.put("bootstrap.servers", "localhost:9092,anotherhost: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", "latest");kafkaParams.put("enable.auto.commit", false);Collection<String> topics = Arrays.asList("topicA", "topicB");final JavaInputDStream<ConsumerRecord<String, String>> stream = KafkaUtils.createDirectStream( streamingContext, LocationStrategies.PreferConsistent(), ConsumerStrategies.<String, String>Subscribe(topics, kafkaParams) );stream.mapToPair( new PairFunction<ConsumerRecord<String, String>, String, String>() { @Override public Tuple2<String, String> call(ConsumerRecord<String, String> record) { return new Tuple2<>(record.key(), record.value()); } })
import java.util.*;
import org.apache.spark.SparkConf;
import org.apache.spark.TaskContext;
import org.apache.spark.api.java.*;
import org.apache.spark.api.java.function.*;
import org.apache.spark.streaming.api.java.*;
import org.apache.spark.streaming.kafka010.*;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;
import scala.Tuple2;
Map<String, Object> kafkaParams = new HashMap<>();
kafkaParams.put("bootstrap.servers", "localhost:9092,anotherhost: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", "latest");
kafkaParams.put("enable.auto.commit", false);
Collection<String> topics = Arrays.asList("topicA", "topicB");
final JavaInputDStream<ConsumerRecord<String, String>> stream =
KafkaUtils.createDirectStream(
streamingContext,
LocationStrategies.PreferConsistent(),
ConsumerStrategies.<String, String>Subscribe(topics, kafkaParams)
);
stream.mapToPair(
new PairFunction<ConsumerRecord<String, String>, String, String>() {
@Override
public Tuple2<String, String> call(ConsumerRecord<String, String> record) {
return new Tuple2<>(record.key(), record.value());
}
})
如需进一步参考,请访问以下链接https://spark.apache.org/docs/latest/streaming-kafka-0-10-integration.html
pes8fvy92#
kafkautils是apachespark流媒体的一部分,而不是apachekafka的一部分org.apache.spark.streaming.kafka.kafkautils网站
2条答案
按热度按时间fykwrbwg1#
以前的kafkautils包是“org.apache.spark.streaming.kafka”。最新的包是“org.apache.spark.streaming.kafka010”。
要设置kafkaparams和主题详细信息,请检查以下代码段,
如需进一步参考,请访问以下链接https://spark.apache.org/docs/latest/streaming-kafka-0-10-integration.html
pes8fvy92#
kafkautils是apachespark流媒体的一部分,而不是apachekafka的一部分
org.apache.spark.streaming.kafka.kafkautils网站