# How to use MR to import nested data into ES？

**URL:** <https://discuss.elastic.co/t/how-to-use-mr-to-import-nested-data-into-es/182114>\
**Category:** Elasticsearch\
**Tags:** es-hadoop\
**Created:** [May 22, 2019, 2:13am UTC](https://discuss.elastic.co/t/how-to-use-mr-to-import-nested-data-into-es/182114 "2019-05-22T02:13:08Z")\
**Posts on this page:** 2\
**Page:** 1

<div class="post-metadata">

**Author:** ![zhentao\_wang](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/zhentao_wang/32/46584_2.png) [@zhentao\_wang](https://discuss.elastic.co/u/zhentao_wang)\
**Post date:** [May 22, 2019, 2:13am UTC](https://discuss.elastic.co/t/how-to-use-mr-to-import-nested-data-into-es/182114/1 "2019-05-22T02:13:08Z")

</div>

**My Trying：**  
public class Hadoop2ElasticsearchWithJSON extends Configured implements Tool {

public static void main(String args) throws Exception {  
int res = ToolRunner.run(new Configuration(), new Hadoop2ElasticsearchWithJSON(), args);  
System.exit(res);  
}

/\*\*

- Map-reduce program for importing Tweet User exampler data.
- Files are named as: User.csv.
- Format:
- nodeId,name
- 1,jack  
\*/  
public int run(String args) throws Exception {  
Configuration conf = this.getConf();conf.setBoolean("mapred.map.tasks.speculative.execution", false);  
conf.setBoolean("mapred.reduce.tasks.speculative.execution", false);  
conf.set("es.nodes", "");  
// conf.set("es.nodes","127.0.0.1:9200");  
conf.set("es.resource", "hadoop-person/hadoop-person");  
conf.set("es.output.json", "true");Job job = Job.getInstance(conf, "es-hadoop-example");  
job.setJarByClass(Hadoop2ElasticsearchWithJSON.class);  
job.setMapperClass(Hadoop2ElasticsearchMapper.class);  
job.setOutputFormatClass(EsOutputFormat.class);  
job.setMapOutputKeyClass(NullWritable.class);  
job.setMapOutputValueClass(MapWritable.class);FileInputFormat.addInputPath(job, new Path("tinkerpop-modern-2.json"));  
FileOutputFormat.setOutputPath(job, new Path("output"));job.waitForCompletion(true);  
return 0;  
}

private static class Hadoop2ElasticsearchMapper extends Mapper\<Object, Text, NullWritable, MapWritable\> {

```auto
public void map(Object key, Text value, Context context) throws IOException, InterruptedException
{
    JSONObject vertex = JSONObject.parseObject(value.toString());
    JSONObject properties = vertex.getJSONObject("properties");
    JSONArray nest = new JSONArray();
    for (String fieldName : properties.keySet()) {
        JSONArray jsonArray = properties.getJSONArray(fieldName);
        for (Object o : jsonArray) {
            JSONObject property = JSONObject.parseObject(o.toString());
            JSONObject jsonObject = new JSONObject();
            jsonObject.put("prop", fieldName);
            jsonObject.put("val", property.getString("value"));
            nest.add(jsonObject);
        }
    }
    MapWritable doc = new MapWritable();
    doc.put(new Text("ontology"), new JsonWritable(nest));
    context.write(NullWritable.get(), doc);
}

```

}  
}

**JsonWritable：**  
/\*\*

Writable representing a JSON object.  
\*/  
public class JsonWritable implements Writable {  
private static final Gson GSON = new Gson();  
private JSONArray jsonArray;

/\*\*

Creates an empty {[@code](https://github.com/code) JsonWritable}.  
\*/  
public JsonWritable() {  
jsonArray = new JSONArray();  
}

public JsonWritable(JSONArray jsonArray) {  
this.jsonArray = jsonArray;  
}

/\*\*

Deserializes a {[@code](https://github.com/code) JsonWritable} object. \*  
[@param](https://github.com/param) in source for raw byte representation  
\*/  
[@override](https://github.com/override)  
public void readFields(DataInput in) throws IOException {  
int cnt = in.readInt();  
byte buf = new byte[cnt];  
in.readFully(buf);  
jsonArray = JSONArray.parseArray(new String(buf, "UTF-8"));  
}

/\*\*

Serializes this object. \*  
[@param](https://github.com/param) out where to write the raw byte representation  
\*/  
[@override](https://github.com/override)  
public void write(DataOutput out) throws IOException {  
byte buf = GSON.toJson(jsonArray).getBytes();  
out.writeInt(buf.length);  
out.write(buf);  
}

/\*\*

Returns the serialized representation of this object as a byte array. \*  
[@return](https://github.com/return) byte array representing the serialized representation of this object  
[@throws](https://github.com/throws) IOException  
\*/  
public byte serialize() throws IOException {  
ByteArrayOutputStream bytesOut = new ByteArrayOutputStream();  
DataOutputStream dataOut = new DataOutputStream(bytesOut);  
write(dataOut);

```auto
return bytesOut.toByteArray();

```

}

public JSONArray getJsonArray() {  
return jsonArray;  
}

[@override](https://github.com/override)  
public String toString() {  
return getJsonArray().toString();  
}

public static JsonWritable create(DataInput in) throws IOException {  
JsonWritable json = new JsonWritable();  
json.readFields(in);

```auto
return json;

```

}

public static JsonWritable create(byte bytes) throws IOException {  
return create(new DataInputStream(new ByteArrayInputStream(bytes)));  
}  
}

---

<div class="post-metadata">

**Author:** ![system](https://us1.discourse-cdn.com/elastic/original/3X/1/a/1ac57faf039f6b580b3f104ef42a2a89e41014de.png) [@system](https://discuss.elastic.co/u/system)\
**Post date:** [June 19, 2019, 2:13am UTC](https://discuss.elastic.co/t/how-to-use-mr-to-import-nested-data-into-es/182114/2 "2019-06-19T02:13:13Z")

</div>

Closed on June 19, 2019, 2:13am UTC

This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.