使用Elasticsearch-hoop-map-reduce将json从HDFS写入到Elasticsearch

Question

我们有一些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属性的值。目前它是由json中的属性index_name提供的。

如果json包含类似的数组类型的属性

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

我们如何将es.resource.write配置为以第一个tag值为例？

我们尝试使用CCD_ 5和CCD_。

另一个问题是，如何使作业索引tags属性的两个值中的json文档？

Answer 1

我们通过执行以下解决了这两个问题

1-在运行方法中，我们将es.resource.write的值设置为以下

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

2-在map函数中，我们使用gson库将json转换为对象

Object currentValue = gson.fromJson(jsonString, Object.class);

• 这里的对象是我们拥有的json的POJO

3-我们可以从Object中提取我们想要的标签，并将其值作为新属性添加到json中。

前面的步骤解决了第一个问题。关于第二个问题（如果我们希望根据标签的数量将同一个json存储到多个索引中），我们只需遍历json中的标签，并更改我们添加的标签属性，然后再次将json传递给收集器。以下是此步骤所需的代码。

@Override
        public void map(LongWritable arg0, Text value, OutputCollector<LongWritable, Text> output, Reporter reporter)
                throws IOException {
            List<String> tags = getTags(value.toString());
            for (String tag : tags) {
                String newJson = value.toString().replaceFirst("\{", "{"tag":""+tag+"",");
                output.collect(arg0, new Text(newJson));
            }
        }