Not possible to query and create indexes in the same Job

[tzolov]

Because the Configuration allows only one es.resource setting it is impossible to query one index and create a different one in the same Job. (see #18)

[costin]

That's actually by design - why would you want to read and write in the same job to ES? You don't gain anything and actually it's quite problematic (though I might be wrong here until the implementation is finished) for parallel querying/insertion.

[ash211]

Suppose I had a workflow that pre-computed an attribute on each
Elasticsearch document and I wanted to save it back to that document?

One example I can think of is you have a large amount of stock purchases
with (unitCount, unitPrice) pairs and you want to multiply those 2 together
to get (unitCount, unitPrice, totalPrice) tuples. You'd want to read and
write from the same index there I think.

This should probably be done by whatever puts data into ES in the first
place, but being able to go back and enrich previously-existing data with a
Hadoop job is a powerful capability.

On Thu, Apr 18, 2013 at 11:13 AM, Costin Leau notifications@github.comwrote:

That's actually by design - why would you want to read and write in the
same job to ES? You don't gain anything and actually it's quite problematic
(though I might be wrong here until the implementation is finished) for
parallel querying/insertion.

—
Reply to this email directly or view it on GitHubhttps://github.com//issues/26#issuecomment-16593629
.

[tzolov]

I have couple of integration tests that (should) read and write from ES in a single Job. The flow looks like this: (1) Read from ES, (2) aggregate in HDP and write the statistics back to ES. Here is the source:
https://github.com/tzolov/elasticsearch-hadoop/tree/master/src/test/java/org/elasticsearch/hadoop/integration/crunch/writable/e2e

I might be missing the details but isn't the parallelism (on the map side) defined by the number of InputSplits? This is irrelevant to number of Jobs. Or perhaps you have mean something else?

[costin]

A Hadoop job is made up of a Mapper/Reducer . In the cases you mentioned, read and write you end up doing two operations so with two Hadoop jobs.
This is not apparent when one uses Cascading/Crunch because the resolution process happens behind the scenes.
Even for cases like Map/Reduce/Reduce (quite common) you actually end up with Map/Reduce, IdentityMapper/Reduce.

As for reading and writing data to the same index, in the same job, considering it's per shard, there are some consistency considerations that need to be taken into account.

[tzolov]

I'm familiar with Hadoop and M/R. Also as Crunch committer I have certain understanding of what's going behind the scenes. Although complex pipelines can produce multi-jobs execution plans this is not the case with the example above. It uses a single Hadoop Job (1 mapper + 1 reducer) to query from ES, aggregate data and write the result in (possibly different) ES index. Standalone version of the sample app: https://gist.github.com/tzolov/5429016

I'm still trying to understand what makes it difficult or unworthy to support this use case.

In the https://github.com/tzolov/elasticsearch-hadoop master branch the issue is solved by introducing ES_QUERY along the ES_RESOURCE. Former is used hold the ES query and later keeps the target index. The ESRecordReader.init() checks for ES_QUERY first and falls back to ES_RESOURCE (to support the existing semantic).
This patch works fine for me. Do you see any problem with this approach?

[tzolov]

To support this feature (e.g. create and update indices in one job) in addition to the patches mentioned above one has to patch also BufferedRestClient as follow:

    public BufferedRestClient(Settings settings) {
        this.client = new RestClient(settings);
        String tempIndex = settings.getTargetResource();
        if (tempIndex == null) {
            // FIX for issue #26 -- start
            tempIndex = settings.getProperty(ConfigurationOptions.ES_QUERY);
            if (tempIndex == null) {
              tempIndex = "";  
            }
            // FIX for issue #26 -- end
        }
     ...