我们将一些json数据存储到hdfs中，并尝试使用elasticsearch hadoop map reduce将数据摄取到elasticsearch中。
我们使用的代码非常简单（如下）

public class TestOneFileJob extends Configured implements Tool {

    public static class Tokenizer extends MapReduceBase
            implements Mapper<LongWritable, Text, LongWritable, Text> {

        @Override
        public void map(LongWritable arg0, Text value, OutputCollector<LongWritable, Text> output,
                Reporter reporter) throws IOException {

            output.collect(arg0, value);
        }

    }

    @Override
    public int run(String[] args) throws Exception {

        JobConf job = new JobConf(getConf(), TestOneFileJob.class);

        job.setJobName("demo.mapreduce");
        job.setInputFormat(TextInputFormat.class);
        job.setOutputFormat(EsOutputFormat.class);
        job.setMapperClass(Tokenizer.class);
        job.setSpeculativeExecution(false);

        FileInputFormat.setInputPaths(job, new Path(args[1]));

        job.set("es.resource.write", "{index_name}/live_tweets");

        job.set("es.nodes", "els-test.css.org");

        job.set("es.input.json", "yes");
        job.setMapOutputValueClass(Text.class);

        JobClient.runJob(job);

        return 0;
    }

    public static void main(String[] args) throws Exception {
        System.exit(ToolRunner.run(new TestOneFileJob(), args));
    }
}

这个代码工作得很好，但我们有两个问题。
第一个问题是价值 es.resource.write 财产。目前由物业提供 index_name 来自json。
如果json包含array类型的属性

{
"tags" : [{"tag" : "tag1"}, {"tag" : "tag2"}]
}

我们如何配置 es.resource.write 抢先一步 tag 例如，值？
我们试着用 {tags.tag} 以及 {tags[0].tag} 但两者都不起作用。
另一个问题是，如何使作业索引为tags属性的两个值中的json文档？