在 Spark 任务中将数据保存到 ElasticSearch答案

【问题标题】：Saving data to ElasticSearch in Spark task在 Spark 任务中将数据保存到 ElasticSearch
【发布时间】：2017-11-28 08:58:52
【问题描述】：

在通过 Kafka 和 Spark 处理 Avro 消息流时，我将处理后的数据保存为 ElasticSearch 索引中的文档。这是代码（简化）：

    directKafkaStream.foreachRDD(rdd ->{

        rdd.foreach(avroRecord -> {
            byte[] encodedAvroData = avroRecord._2;
            MyType t = deserialize(encodedAvroData);

    // Creating the ElasticSearch Transport client
    Settings settings = Settings.builder()
            .put("client.transport.ping_timeout", 5, TimeUnit.SECONDS).build();
    TransportClient client = new PreBuiltTransportClient(settings)
            .addTransportAddress(new TransportAddress(InetAddress.getByName("localhost"), 9300));

    IndexRequest indexRequest = new IndexRequest("index", "item", id)
            .source(jsonBuilder()
                    .startObject()
                    .field("name", name)
                    .field("timestamp", new Timestamp(System.currentTimeMillis()))
                    .endObject());

    UpdateRequest updateRequest = new UpdateRequest("index", "item", id)
            .doc(jsonBuilder()
                    .startObject()
                    .field("name", name)
                    .field("timestamp", new Timestamp(System.currentTimeMillis()))
                    .endObject())
            .upsert(indexRequest);

    client.update(updateRequest).get();

    client.close();

一切都按预期进行；唯一的问题是性能：保存到 ES 需要一些时间，我想这是因为我为每个 RDD 打开/关闭了一个 ES 传输客户端。 Spark documentation 建议这种方法非常正确：据我了解，唯一可能的优化是使用 rdd.foreachPartition，但我只有一个分区，所以我不确定这是否有益。还有其他解决方案可以实现更好的性能吗？

【问题讨论】：

只是出于好奇——为什么不使用 elasticsearch-hadoop？

标签： apache-kafka spark-streaming elasticsearch-java-api spark-avro

【解决方案1】：

因为每当处理 RDD 的记录时，您都会创建新的连接。所以，我认为使用foreachPartition 将获得更好的性能，而不管只有一个分区，因为它可以帮助您将 ES 连接实例带到外部，在循环中重用它。

【讨论】：

【解决方案2】：

我会将处理后的消息流式传输回一个单独的 Kafka 主题，然后使用 Kafka Connect 将它们发送到 Elasticsearch。这将您的 Spark 特定处理与将数据导入 Elasticsearch 分离。

实际操作示例：https://www.confluent.io/blog/blogthe-simplest-useful-kafka-connect-data-pipeline-in-the-world-or-thereabouts-part-2/

【讨论】：