KafkaUtils.createDirectStream 不采用正确的参数 - Spark Streaming + Kafka答案

【问题标题】：KafkaUtils.createDirectStream does not take right parameters - Spark Streaming + KafkaKafkaUtils.createDirectStream 不采用正确的参数 - Spark Streaming + Kafka
【发布时间】：2018-06-20 17:02:11
【问题描述】：

我有一个将序列化的 Twitter 数据发送到 Kafka 主题的应用程序。到目前为止一切顺利。

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

方法createDirectStream(JavaStreamingContext, Class -K-, 类-V-、类-KD-、类-VD-、映射-String、String-、Set-String-) KafkaUtils 类型不适用于参数（JavaStreamingContext、Class-String-、Class-Status-、类-StringDeserializer-，类-StatusDeserializer-， Map-String,String-,Set-String-)

检查 Spark Javadoc，我的参数对我来说似乎仍然正确。

我的代码是：

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) {


    }

}

【问题讨论】：

标签： java apache-spark apache-kafka spark-streaming twitter4j

【解决方案1】：

看起来问题出在“stream_data.StatusDeserializer.class”上。你能请这个自定义反序列化器类的代码吗？另外，请你看看这个Kafka Consumer for Spark written in Scala for Kafka API 0.10: custom AVRO deserializer。

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

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

【讨论】：