如何将javainputdstream json转换为elasticsearch java

zrfyljdw 于 2021-06-08 发布在 Kafka

关注(0)|答案(2)|浏览(340)

嗨，伙计们，我和Kafka一起工作>spark streaming>elasticsearch。但我不会将spark streaming javainputdstream json转换为elasticsearch。
我的代码：

SparkConf conf = new SparkConf()
            .setAppName("Streaming")
            .setMaster("local")
            .set("es.nodes","localhost:9200")
            .set("es.index.auto.create","true");
    JavaStreamingContext streamingContext = new JavaStreamingContext(conf, new Duration(5000));
    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", "exastax");
    kafkaParams.put("auto.offset.reset", "latest");
    kafkaParams.put("enable.auto.commit", false);

    Collection<String> topics = Arrays.asList("loglar");
    JavaInputDStream<ConsumerRecord<String, String>> stream =
            KafkaUtils.createDirectStream(
                    streamingContext,
                    LocationStrategies.PreferConsistent(),
                    ConsumerStrategies.<String, String>Subscribe(topics, kafkaParams)
            );

    JavaPairDStream<String, String> finisStream = stream.mapToPair(record -> new Tuple2<>("", record.value()));
    finisStream.print();
    JavaEsSparkStreaming.saveJsonToEs(finisStream,"spark/docs");
    streamingContext.start();
    streamingContext.awaitTermination();

}

javaessparkstreaming.savejsontoes（finisstream，“spark/docs”）；>>FinishStream无法工作，因为它不是javadstream。如何转换javadstream？

elasticsearch apache-kafka apache-spark spark-streaming

来源：https://stackoverflow.com/questions/45507324/how-to-javainputdstream-json-to-elasticsearch-java

2条答案

按热度按时间

yduiuuwa1#

JavaEsSparkStreaming.saveJsonToEs 使用
JavaDStream JavaEsSparkStreaming.saveToEsWithMeta 使用 JavaPairDStream 要修复代码：

JavaDStream<String> finisStream = stream.map(new Function<Tuple2<String, String>, String>() {
    public String call(Tuple2<String, String> stringStringTuple2) throws Exception {
        return stringStringTuple2._2();
    }
});

JavaEsSparkStreaming.saveJsonToEs(finisStream,"");

赞(0）回复(0）举报 2021-06-08

dtcbnfnu2#

答案太多了！但我破解了这个密码：

JavaDStream<String> stream1 = stream.map(
                new Function<ConsumerRecord<String, String>, String>() {
                    @Override
                    public String call(ConsumerRecord<String, String> r) {
                        return r.value();
                    }
                }
        );
           JavaEsSparkStreaming.saveJsonToEs(stream1,"spark/docs");

赞(0）回复(0）举报 2021-06-08

我来回答

如何将javainputdstream json转换为elasticsearch java

2条答案

相关问题

热门标签

最新问答