Hadoop HDFS MapReduce 输出到 MongoDb答案

【问题标题】：Hadoop HDFS MapReduce output into MongoDbHadoop HDFS MapReduce 输出到 MongoDb
【发布时间】：2013-10-21 16:06:04
【问题描述】：

我想编写 Java 程序，它从 HDFS 读取输入，使用 MapReduce 处理它并将输出写入 MongoDb。

这是场景：

我有一个 Hadoop 集群，它有 3 个数据节点。
Java 程序从 HDFS 读取输入，使用 MapReduce 对其进行处理。
最后，将结果写入 MongoDb。

实际上，从 HDFS 读取并使用 MapReduce 进行处理很简单。但是我被困在将结果写入 MongoDb 中。是否有任何 Java API 支持将结果写入 MongoDB？另一个问题是，既然是 Hadoop Cluster，所以我们不知道哪个 datanode 会运行 Reducer 任务并生成结果，是否可以将结果写入安装在特定服务器上的 MongoDb 中？

如果我想将结果写入HDFS，代码是这样的：

@Override
public void reduce(Text key, Iterable<LongWritable> values, Context context) throws IOException, InterruptedException 
{
    long sum = 0;
    for (LongWritable value : values) 
    {
        sum += value.get();
    }

    context.write(new Text(key), new LongWritable(sum));
}

现在我想将结果写入 MongoDb 而不是 HDFS，我该怎么做？

【问题讨论】：

你是什么意思，你在将结果写入 MongoDB 时遇到了困难？ MongoDB 有一个可用的 Java API (docs.mongodb.org/ecosystem/drivers/java)。
在 Reducer 类中使用 mongo-java-driver？会成功吗？
为什么不试试呢？（为什么要将数据从 HDFS 复制到 MongoDB？）
大概能查询到吧。疯了，我知道。
嘿@user2597504，你能告诉我它是如何工作的（或者你是如何解决这个问题的）吗？你的问题和我好奇的一模一样。

标签： java mongodb hadoop mapreduce hdfs

【解决方案1】：

你想要«MongoDB Connector for Hadoop»。 examples。

很容易在 Reducer 中添加代码，作为副作用，将数据插入数据库。避免这种诱惑。使用连接器而不是仅仅插入数据作为 reducer 类的副作用的一个原因是推测性执行：Hadoop 有时可以并行运行两个完全相同的 reduce 任务，这可能导致无关的插入和重复数据。

【讨论】：

添加了更多关于为什么使用连接器库是个好主意的细节。

【解决方案2】：

是的。你像往常一样写信给 mongo。您的 mongo db 设置为在分片上运行这一事实是对您隐藏的细节。

【讨论】：

【解决方案3】：

我花了一上午的时间来实现相同的场景。这是我的解决方案：

创建三个类：

Experiment.java： 用于作业配置和提交
MyMap.java： 映射器类

MyReduce.java：减速器类

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import com.mongodb.hadoop.io.BSONWritable;
import com.mongodb.hadoop.mapred.MongoOutputFormat;

public class Experiment extends Configured implements Tool{

     public int run(final String[] args) throws Exception {
        final Configuration conf = getConf();
        conf.set("mongo.output.uri", args[1]);

        final JobConf job = new JobConf(conf);

        FileInputFormat.setInputPaths(job, new Path(args[0]));
        job.setJarByClass(Experiment.class);

        job.setInputFormat(org.apache.hadoop.mapred.TextInputFormat.class);
        job.setMapperClass(MyMapper.class);
        job.setReducerClass(MyReducer.class);
        job.setOutputFormat(MongoOutputFormat.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(BSONWritable.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);

        JobClient.runJob(job);

        return 0;
    }

    public static void main(final String[] args) throws Exception{

        int res = ToolRunner.run(new TweetPerUserToMongo(), args);
        System.exit(res);
    }
}

当您从集群运行 Experiment 类时，您将输入两个参数。第一个参数是来自 HDFS 位置的输入源，第二个参数是指将保留结果的 mongodb URI。这是一个示例调用。假设您的 Experiment.java 位于包名 org.example 下。

sudo -u hdfs hadoop jar ~/jar/myexample.jar org.example.Experiment myfilesinhdfs/* mongodb://192.168.0.1:27017/mydbName.myCollectionName

这可能不是最好的方法，但它对我有用。

【讨论】：