使用 java 将 rdd 保存到 mongo 数据库

Question

我正在尝试使用 Java 在 MongoDB 中保存推文，这就是我所拥有的；

JavaStreamingContext ssc = new JavaStreamingContext(sc, new Duration(
            3000));
    JavaDStream<Status> tweets = TwitterUtils.createStream(ssc);

    JavaDStream<String> statuses = tweets
            .map(new Function<Status, String>() {
                public String call(Status status) {
                    return status.getUser().getName() + ":"
                            + status.getText();
                }
            });

    JavaDStream<String> users = tweets.map(new Function<Status, String>() {
        public String call(Status status) {
            return status.getUser().getName();
        }
    });


    users.foreachRDD(new Function<JavaRDD<String>, Void>() {
        public Void call(JavaRDD<String> rdd) throws Exception {
            if (rdd.count() > 0)
                rdd.saveAsTextFile("storage/users/test" + rdd.id()
                        + "_.txt");
            return null;
        }
    });

如您所见，我可以使用 rdd.saveAsTextFile 将用户存储在文本文件中，但我需要的是一种将此 rdd 保存到数据库 (MongoDB) 的方法。

Answer 1

您可以使用 MongoDB Hadoop Connector 将 RDD 存储到 MongoDB 使用 com.mongodb.hadoop.MongoOutputFormat:

Configuration config = new Configuration();
config.set("mongo.output.format", "com.mongodb.hadoop.MongoOutputFormat");
config.set("mongo.output.uri", "mongodb://host:port/database.collection");
rdd.saveAsNewAPIHadoopFile("file://this-is-not-used",
                           <keyClass>,
                           <valueClass>,
                           MongoOutputFormat.class,
                           config);

查看 an example project 了解如何执行此操作可能也会有所帮助。

使用 java 将 rdd 保存到 mongo 数据库

Save rdd into mongo database using java

mongodb

hadoop

spark-streaming

rdd