mongodb - データをHiveからMongoDBに移動する最も効率的な方法は何ですか？

Question

データをHiveからMongoDBに移動するためのエレガントで簡単かつ高速な方法はありますか？

score 2 · Accepted Answer

Hadoop-MongoDBコネクタを使用してエクスポートを実行できます。ジョブのmainメソッドでHiveクエリを実行するだけです。この出力は、データをに挿入するためにマッパーによって使用されMongoDBます。

例：

ここでは、単純なHiveクエリを使用して、セミコロンで区切られたテキストファイル（id; firstname; lastname ）をMongoDBコレクションに挿入しています。

import java.io.IOException;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;
import java.sql.Statement;

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.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

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

public class HiveToMongo extends Configured implements Tool {

    private static class HiveToMongoMapper extends
            Mapper<LongWritable, Text, IntWritable, BSONWritable> {

        //See: https://issues.apache.org/jira/browse/HIVE-634
        private static final String HIVE_EXPORT_DELIMETER = '\001' + "";
        private IntWritable k = new IntWritable();
        private BSONWritable v = null;

        @Override
        public void map(LongWritable key, Text value, Context context) 
          throws IOException, InterruptedException {

            String [] split = value.toString().split(HIVE_EXPORT_DELIMETER);

            k.set(Integer.parseInt(split[0]));
            v = new BSONWritable();
            v.put("firstname", split[1]);
            v.put("lastname", split[2]);
            context.write(k, v);

        }
    }

    public static void main(String[] args) throws Exception {
        try {
            Class.forName("org.apache.hadoop.hive.jdbc.HiveDriver");
        }
        catch (ClassNotFoundException e) {
            System.out.println("Unable to load Hive Driver");
            System.exit(1);
        }

        try {
            Connection con = DriverManager.getConnection(
                "jdbc:hive://localhost:10000/default");

            Statement stmt = con.createStatement();    
            String sql = "INSERT OVERWRITE DIRECTORY " +
                    "'hdfs://localhost:8020/user/hive/tmp' select * from users";
            stmt.executeQuery(sql);

        }
        catch (SQLException e) {
            System.exit(1);
        }

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

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

        Configuration conf = getConf();
        Path inputPath = new Path("/user/hive/tmp");
        String mongoDbPath = "mongodb://127.0.0.1:6900/mongo_users.mycoll";
        MongoConfigUtil.setOutputURI(conf, mongoDbPath);

        /*
        Add dependencies to distributed cache via 
        DistributedCache.addFileToClassPath(...) :
        - mongo-hadoop-core-x.x.x.jar
        - mongo-java-driver-x.x.x.jar
        - hive-jdbc-x.x.x.jar
        HadoopUtils is an own utility class
        */
        HadoopUtils.addDependenciesToDistributedCache("/libs/mongodb", conf);
        HadoopUtils.addDependenciesToDistributedCache("/libs/hive", conf);

        Job job = new Job(conf, "HiveToMongo");

        FileInputFormat.setInputPaths(job, inputPath);
        job.setJarByClass(HiveToMongo.class);
        job.setMapperClass(HiveToMongoMapper.class);
        job.setInputFormatClass(TextInputFormat.class);
        job.setOutputFormatClass(MongoOutputFormat.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);
        job.setNumReduceTasks(0);

        job.submit();
        System.out.println("Job submitted.");
        return 0;
    }
}

1つの欠点は、中間のHive出力を格納するために「ステージング領域」（/ user / hive / tmp）が必要になることです。さらに、私が知る限り、Mongo-Hadoopコネクタはアップサートをサポートしていません。

よくわかりませんが、Hive実行せずにデータをフェッチしhiveserverて、Thriftサービスを公開することもできます。これにより、オーバーヘッドをいくらか節約できます。org.apache.hadoop.hive.cli.CliDriver#processLine(String line, boolean allowInterupting)実際にクエリを実行するHiveのメソッドのソースコードを見てください。次に、次のようなものを一緒にハックできます。

...
LogUtils.initHiveLog4j();
CliSessionState ss = new CliSessionState(new HiveConf(SessionState.class));
ss.in = System.in;
ss.out = new PrintStream(System.out, true, "UTF-8");
ss.err = new PrintStream(System.err, true, "UTF-8");
SessionState.start(ss);

Driver qp = new Driver();
processLocalCmd("SELECT * from users", qp, ss); //taken from CliDriver
...

サイドノート：

また、チェックする可能性のあるハイブモンゴコネクタの実装もあります。に同様のコネクタを実装する場合は、 Hive-HBaseコネクタの実装を確認してアイデアを得る価値もありますMongoDB。

score 1 · Accepted Answer

Sqoopを調べましたか？これにより、HadoopデータベースとSQL/NoSQLデータベース間でのデータの移動が非常に簡単になります。この記事では、Hiveでの使用例も示しています。

score 1 · Accepted Answer

hadoop-MongoDBコネクタプロジェクトを見てください：

http://api.mongodb.org/hadoop/MongoDB%2BHadoop+Connector.html

「この接続は、MongoDBデータをHadoopに読み込むこと（MapReduceジョブやHadoopエコシステムの他のコンポーネントで使用するため）と、Hadoopジョブの結果をMongoDBに書き出すことの両方を可能にするという形をとります。」

ユースケースで機能するかどうかはわかりませんが、一見の価値があります。

mongodb - データをHiveからMongoDBに移動する最も効率的な方法は何ですか？

3 に答える 3

Related

Reference