CreateDirectStream不接受正确的参数-Spark Streaming+Kafka

Question

我有一个应用程序，它发送序列化的Twitter数据到一个Kafka主题。到目前为止一切都很好。

使用者应用程序应该读取数据并对其进行反序列化。现在，当我调用kafkautils.createDirectStream时，我认为我放入了正确的参数（正如您在抛出的错误中看到的那样），所以我不明白为什么它不能工作。

KafkaUtils类型中的方法createDirectStream（JavaStreamingContext,class-k-,class-v-,class-kd-,class-vd-,Map-String,string-,set-string-）不适用于参数（JavaStreamingContext,class-string-,class-status-,class-string-,class-string,string-,set-string-）

检查Spark Javadoc时，我的params对我来说仍然是正确的。

我的代码是：

Set<String> topics = new HashSet<>();
topics.add("twitter-test");
JavaStreamingContext jssc = new JavaStreamingContext(jsc, new Duration(duration));
Map<String, String> props = new HashMap<>();
//some properties...
JavaPairInputDStream messages =  KafkaUtils.createDirectStream(jssc, String.class, Status.class, org.apache.kafka.common.serialization.StringDeserializer.class, stream_data.StatusDeserializer.class, props, topics);

状态序列化程序代码：

public class StatusSerializer implements Serializer<Status> {

  @Override public byte[] serialize(String s, Status o) {

           try {
                ByteArrayOutputStream baos = new ByteArrayOutputStream();
                ObjectOutputStream oos = new ObjectOutputStream(baos);
                oos.writeObject(o);
                oos.close();
                byte[] b = baos.toByteArray();
                return b;
            } catch (IOException e) {
                return new byte[0];
            }
        }

      @Override public void close() {

      }

    @Override
    public void configure(Map<String, ?> configs, boolean isKey) {


    }

}

Answer 1

看起来问题出在“stream_data.statusDeserializer.class”上。你能请这个自定义反序列化程序类的代码。另外，请您看看这个用Scala for Kafka API0.10编写的Kafka Spark消费者：自定义AVRO反序列化器。

在KafkaParam参数中包括以下内容。

key.deserializer -> classOf[StringDeserializer]
value.deserializer -> classOf[StatusDeserializer]