Support UPDATE output mode for Spark Structured Streaming

[jbaiera]

We now support Spark Structured Streaming as of 6.x, but only in APPEND mode. It would be beneficial to support UPDATE mode, with the requirement that the user MUST provide a field to be used for a document ID.

[kant111]

can we make this backward compatible also? That way users can use this feature along with ES >= 5.5 or something

[jbaiera]

While the project's official stance is that we maintain BWC with the previous major's most recent minor release, we give it our best effort to maintain backwards compatibility with as many versions of Elasticsearch as we can.

[kant111]

Any sort of timeline or release number on this?

[jbaiera]

We usually do not commit to timelines or release targets on public issues.

[kant111]

@jbaiera can I use update mode and forEachSink to write to ES for now?

[jbaiera]

That seems like a reasonable work around in the mean time.

[kant111]

@jbaiera Thanks. Can you please provide some more details on how to do that? For example, should I use elastic-hadoop connector or elastic search java driver? Any examples because I dont see in docs on how to do this?

I am trying to write streaming dataset to elastic search in update mode.

currently using Spark 2.3.0 and E.S 5.3.2

[jbaiera]

This very much depends on your use case. In my experience Spark does not offer much in terms of solid documentation for writing your own sinks. Much of the current implementation for ES-Hadoop was the product of a month of trial and error and reverse engineering existing sink implementations. If you don't much care about the expectations around transaction acknowledgement and skipping acked transactions during a recovery, then the basic Foreach sink should be useable for your case.

I understand that the current java rest client can sometimes cause issues in Hadoop/Spark environments because of clashing dependencies. ES-Hadoop has its own Rest Client implementation that it uses to get around this. I don't recommend using the ES-Hadoop internals to build this as they are not covered by our semantic versioning scheme, and may break between any release with no notice. Because of this, I recommend using the available clients for Elasticsearch if you are going to build something as a work around.

[kant111]

@jbaiera 1) I don't see any REST client for Elastic search 5.3.2. 2) when I used the BulkProcessor from the transport java driver to write to ES I get task not serializable exception.

Here is the code

public class EsSink extends ForeachWriter<Row> {

    private TransportClient client;
    private BulkProcessor bulkProcessor;

    public EsSink(String cluster, String host, int port) throws UnknownHostException {
        Settings settings = Settings.builder()
                .put("cluster.name", cluster).build();
        String[] elasticSearchIps = host.split(",");
        InetSocketTransportAddress[] inetSocketTransportAddresses = new InetSocketTransportAddress[elasticSearchIps.length];
        for (int i = 0; i < elasticSearchIps.length; i++) {
            inetSocketTransportAddresses[i] = new InetSocketTransportAddress(InetAddress.getByName(elasticSearchIps[i]), port);
        }
        this.client = new PreBuiltTransportClient(settings)
                .addTransportAddresses(inetSocketTransportAddresses);

        this.bulkProcessor = BulkProcessor.builder(
                client,
                new BulkProcessor.Listener() {
                    @Override
                    public void beforeBulk(long executionId,
                                           BulkRequest request) {}

                    @Override
                    public void afterBulk(long executionId,
                                          BulkRequest request,
                                          BulkResponse response) {}

                    @Override
                    public void afterBulk(long executionId,
                                          BulkRequest request,
                                          Throwable failure) {}
                })
                .setBulkActions(10000)
                .setBulkSize(new ByteSizeValue(5, ByteSizeUnit.MB))
                .setFlushInterval(TimeValue.timeValueSeconds(5))
                .setConcurrentRequests(1)
                .setBackoffPolicy(
                        BackoffPolicy.exponentialBackoff(TimeValue.timeValueMillis(100), 3))
                .build();
    }

    @Override
    public boolean open(long l, long l1) {
        return true;
    }

    @Override
    public void process(Row row) {
        String[] fieldNames = row.schema().fieldNames();
        Seq<String> fieldNamesSeq = JavaConverters.asScalaIteratorConverter(Arrays.asList(fieldNames).iterator()).asScala().toSeq();
        String jsonDocument = row.getValuesMap(fieldNamesSeq).toString();
        IndexRequest indexRequest = Requests.indexRequest("hello").type("foo").id(row.get("id")).source(jsonDocument, XContentType.JSON);
        this.bulkProcessor.add(indexRequest);
    }

    @Override
    public void close(Throwable throwable) {
        this.bulkProcessor.close();
        this.client.close();
    }
}

[jbaiera]

If you are using the Transport client you will need to construct the client objects lazily and mark them as transient as they cannot be serialized from the driver to the cluster.

[kant111]

ok! got it. Btw, I haven't implemented a custom sink before but I looked at the Es-Hadoop-connector code to see how hard would it be to implement the update mode. I feel like append mode code already establishes a good code structure in terms of implementing the right interfaces and so on. so my question really now is what changes or how big of a change I would need to make if I were to implement the update mode?

[kant111]

The interface below is already implemented in the ES-Hadoop-Connector and output modes are taken care by spark other words spark sends whatever output mode a user sets so It looks to me that we just need to change this one IF but at the same time I feel thats too good to be true?

trait Sink {
  def addBatch(batchId: Long, data: DataFrame): Unit
}

[jbaiera]

@kant111 I'm happy to see that you've taken in interest in the code. That if-statement indeed does block off the option itself from being used, but it doesn't do anything to make sure that the Sink implementation observes the invariants set forth in the UPDATE mode, or follows what a user might expect from UPDATE mode in regards to Elasticsearch. The following things would need to be in place to accept working with UPDATE mode:

First, in UPDATE mode, the underlying connector capabilities should be fine as long as it is using the upsert method of ingestion to ES. Alongside this mode, a field must be marked as an ID in the configuration. To support UPDATE mode, it should be checked that these settings are present, or even to set them in the connector at this point, throwing an error if they have values already that are incompatible.

Second, for us to support UPDATE mode, we need sufficient testing to make sure that the connector is performing the required operations, and can see the expected outcomes. Building these tests can be time consuming, and thus they were left off of the initial implementation.

Finally, there is no plan to support any other output modes than APPEND and (eventually) UPDATE. The COMPLETE output mode (and any future ones) would need to be blocked off since Elasticsearch would not support that kind of output mode.