1

I'm using mongo-hadoop connector which makes that Hadoop can get data from MongoDB and save into MongoDB. I found MongoDB data is duplicated after Hadoop Map-Reduce job with MongoDB data.

environment: Hadoop version is 3.1.1, MongoDB version is 4.0.4, mongo-hadoop connector is mongo-hadoop-core-2.0.2.jar, mongo-java-driver-3.8.2.jar, and Docker version is 18.03.1-ce. There are two local server named server1, server2. They have public IP so I can make hadoop cluster or mongo sharding environment on them. To run hadoop and MongoDB, I used Docker and Docker orchestration so that containers can exchange packets each other on an overlay network. server 1 has Hadoop master container, Hadoop slave1 container, MongoDB router container, MongoDB configurator container, and server 2 has Hadoop slave2 container, Hadoop slave3 container, and MongoDB shard1 container, MongoDB shard2 container. MongoDB has 30MB tsv data file and chunk size is 8MB. When I setup Hadoop clustering and MongoDB sharding, I let the containers connect each other by their container name(ex, Hadoop master sends ping with 'slave1' or 'mongorouter'.. and it works well).

problem : The problem is, after setting Hadoop cluster and MongoDB Sharding(the sharded collection is just one) WordCount MR job using MongoDB data(the 30MB data), the WordCount result is duplicated, in detail, the data is multiplied by 2. For example, the normal result is [a 1, b 2] then the duplicated result is [a 2, b 4]. If I make another sharded collection(same 30MB data, same code, same database, just different collections name, so the sharded collections are two. MR job uses only the new sharded collection) then the result is multiplied by 3( [a 3, b 6] ). If I added more sharded collections the same way, the result is multiplied proportionally. If I don't set the Mongo sharding environment, the result is what I expect. I really don't know what is happening. I noticed MongoCollectionSpliter is increasing. The more sharded collections, the more MongoCollectionSpliter in the MR result(I attached the logs below)

This is the WordCount MR code.

import java.io.IOException;
import java.util.StringTokenizer;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.bson.BSONObject;
import org.bson.BasicBSONObject;

import com.mongodb.hadoop.MongoInputFormat;
import com.mongodb.hadoop.MongoOutputFormat;
import com.mongodb.hadoop.io.BSONWritable;
import com.mongodb.hadoop.util.MongoConfigUtil;

public class BigdataBench{
    public static void main(String[] args) throws Exception {
            Configuration conf = new Configuration();

            MongoConfigUtil.setInputURI(conf, "mongodb://" + args[0]);
            MongoConfigUtil.setOutputURI(conf, "mongodb://" + args[1]);

            Job job = Job.getInstance(conf, "WordCount");

            job.setJarByClass(BigdataBench.class);

            job.setMapperClass(Map.class);
            job.setReducerClass(Reduce.class);

            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(IntWritable.class);

            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(BSONWritable.class);

            job.setInputFormatClass(MongoInputFormat.class);
            job.setOutputFormatClass(MongoOutputFormat.class);

            job.waitForCompletion(true);
    }

    public static class Map extends Mapper<Object, BSONObject, Text, IntWritable> {
            private final static IntWritable one = new IntWritable(1);
            private final Text dataOutput = new Text();

            public void map(Object key, BSONObject value, Context context) throws IOException, InterruptedException {
                    String data = value.get("data").toString();

            for(String whiteSpaceSplit : data.split(" ")) {
                    String[] tapSplit = whiteSpaceSplit.split("\t");

                    for(String split : tapSplit) {
                            dataOutput.set(split);
         context.write(dataOutput, one);
                    }
            }

            }
    }

    public static class Reduce extends Reducer<Text, IntWritable, Text, BSONWritable> {
            private BSONWritable reduceResult = new BSONWritable();

            public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
                    BasicBSONObject output = new BasicBSONObject();

                    int sum = 0;
                    for(IntWritable s : values) {
                            sum+=s.get();
                    }

                    String wordCount = String.valueOf(sum);

                    output.put("word", wordCount);
                    reduceResult.setDoc(output);
                    context.write(key, reduceResult);
            }
    }
}

I arranged my problem as a matrix. image

The MR results are too long so I put them into this google doc.

If you have the same problem or solutions, please answer and help me. Thanks in advance.

Your Answer

By clicking "Post Your Answer", you acknowledge that you have read our updated terms of service, privacy policy and cookie policy, and that your continued use of the website is subject to these policies.

Browse other questions tagged or ask your own question.