阿里云MongoDB与EMR的HelloWorld

越来越多的应用采用MongoDB作为数据存储层，性能高，扩展性强，通过WriteCocern参数还可以控制写入持久级别，CAP上灵活配置。文档型的存储结构又是特别适合物联网，游戏等领域，这些数据也蕴藏这巨大的价值，就像是金矿一样，需要挖掘。虽然MongoDB提供了MapReduce功能，但功能相对薄弱，如果说MongoDB MapReduce是铁锹，Spark就是一台真正的挖掘机。

阿里云云数据库已经推出了MongoDB云服务，EMR（E-MapReduce）也是公测期，EMR提供了便捷的Spark服务，本篇文章将给大家介绍下如何使用使用阿里云服务，构建基于MongoDB的大数据计算平台。

EMR服务申请和创建

准备工作

钱，服务是要买的，学习为目的可以使用小时付费
提前开通OSS，EMR服务是依赖OSS的，所以建议提前开通OSS

申请EMR公测资格

点击申请地址，开通一般是在1-2个工作日左右，目前公测期间EMR服务的价格与ECS保持一致。长期使用可以按月购买，最小规模大概1000元左右，学习的话可以按小时付费，不过用好后请记得释放。

创建EMR集群

申请通过后就可以创建集群了，注意下运行日志的路径，需要指定一个OSS Bucket存放日志，为了方便追踪状态，建议开启。

输入好密码后就可以点击下一步了进行软件配置，默认选择Hadoop集群即可，继续下一步。因为EMR实际上是运行在ECS上，所以需要安全组配置，没有的话需要创建一个。另外，测试目的的话需要最小化集群配置，Core减小到一个节点，生产目的的话强烈建议多个Core。

继续，支付订单，等待集群创建，大概30秒后集群即可创建完毕。在ECS控制台上也可以看到新生产出的两个ECS节点，上面就运行着EMR服务，我们可以像使用普通ECS的方式一样登陆到节点上。

OK，至此Spark集群已经构建完成。

购买阿里云云数据库MongoDB

因为MongoDB已经是商业化的服务，所以正常购买即可，但需要注意的是，一定要购买与EMR服务在同一个可用区的实例，否则网络是不通的。

EMR可用区查看

MongoDB可用区选择

等待30S后查看控制台，MongoDB实例创建成功。

创建好后，先写上几条数据，为后面的DEMO做准备，如图：

检查网络连通性

开始之前还需要检查下EMR与MongoDB云服务的网络连通性，看看是否是畅通的。登陆到EMR创建好的ECS上，通过telnet命令来探测：

telnet dds-xxxxxxx.mongodb.rds.aliyuncs.com 3717

如果发现无法连接有几个可能性逐一排查：

EMR服务与MongoDB云服务不在同一个可用区，阿里云的网络规则下是不通的，需要重新购买
安全组限制了内网进出口，可以登陆ECS控制台修改安全组规则，让其可以访问MongoDB服务端口
由于欠费等原因，生长出来的实例被回收了，也可以通过控制台查看实例状态是否正常

至此，资源都已经Ready，接下来我们一起构建Spark 计算用的Jar包吧。

Spark任务编写

Jar包依赖

要想Spark访问MongoDB，必须找到相对应的Hadoop Connector和相关的Jar包，可以参考如下Maven POM配置。具体的版本，根据自己的实际需要去更新。

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.aliyun.mongodb</groupId>
    <artifactId>spark-test</artifactId>
    <version>1.0-SNAPSHOT</version>

    <build>
       <plugins>
           <plugin>
        <artifactId>maven-assembly-plugin</artifactId>
        <configuration>
            <archive>
                <manifest>
                    <mainClass>fully.qualified.MainClass</mainClass>
                </manifest>
            </archive>
            <descriptorRefs>
                <descriptorRef>jar-with-dependencies</descriptorRef>
            </descriptorRefs>
        </configuration>
    </plugin>
       </plugins>
    </build>

    <dependencies>
        <dependency>
            <groupId>org.mongodb</groupId>
            <artifactId>mongodb-driver</artifactId>
            <version>3.2.2</version>
        </dependency>
        <dependency>
            <groupId>org.mongodb.mongo-hadoop</groupId>
            <artifactId>mongo-hadoop-core</artifactId>
            <version>1.5.2</version>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_2.10</artifactId>
            <version>1.6.0</version>
        </dependency>
    </dependencies>
</project>

Job编写

通过MongoDB控制台准备好MongoDB的几个属性：

两个访问地址，注意，是两个
用户名，密码，从MongoDB上拉取需要读权限，如果还希望数据写回MongoDB，那写权限也需要准备好
MongoDB集群名，以mgset开头

   private static String DEFAULT_AUTH_DB = "admin";

   private static String seed1         = "dds-xxxxx1.mongodb.rds.aliyuncs.com:3717";
   private static String seed2         = "dds-xxxxx2.mongodb.rds.aliyuncs.com:3717";
   private static String username      = "root";
   private static String password      = "123456";
   private static String replSetName   = "mgset-1234567";

接下来构建MongoDB ConnectionURI，具体的规则参考如下代码，参考github文档，或者跟着下面的代码抄写。最终要有三个URI：

mongoURI 用来鉴权
inputURI 数据输入地址
ouputURI 数据输出地址


   private static String authURIPrefix = "mongodb://" +
                                          username + ":" + password + "@" +
                                          seed1 + "," + seed2 + "/";
   private static String authURISuffix = "?replicaSet=" + replSetName;
   private static String inputColl     = "testdb.input";
   private static String outputColl    = "testdb.output";

   private static String mongoURI      = authURIPrefix + DEFAULT_AUTH_DB + authURISuffix;
   private static String inputURI      = authURIPrefix + inputColl + authURISuffix;
   private static String outputURI     = authURIPrefix + outputColl + authURISuffix;

至此，访问环境相关的变量都已经初始化完成，正式进入到Job内容，这里的Demo很简单，不能免俗的Hello World风格，但麻雀虽小五脏俱全，从配置到输入到计算再到输出，完整的一套流程。

首先撞见SparkContext，Spark作业的生命周期都会伴随着这个Context，并且配置Configuration对象，Configuration对象维护着上面提到的访问地址参数，更详细参数说明可以参考github。

JavaSparkContext sc = new JavaSparkContext(new SparkConf());

Configuration config = new Configuration();
config.set("mongo.job.input.format", "com.mongodb.hadoop.MongoInputFormat");
config.set("mongo.job.output.format", "com.mongodb.hadoop.MongoOutputFormat");
config.set("mongo.auth.uri", mongoURI);
config.set("mongo.input.uri", inputURI);
config.set("mongo.output.uri", outputURI);

接下来轮到获取数据RDD了，RDD是Spark中的数据表达形式。这里要注意RDD Value类型，是BSONObject，BSON是MongoDB文档数据的表现形式。通过这样一条语句做了BSON到RDD的映射。

JavaPairRDD<Object, BSONObject> documents = sc.newAPIHadoopRDD(
              config,                       // Configuration
              MongoInputFormat.class,   // InputFormat: read from a live cluster.
              Object.class,             // Key class
              BSONObject.class          // Value class
      );

有了数据，就可以开始计算了，简单的做个mapValues动作，可以注意看，返回的仍然是个RDD，不过这个RDD是经过map动作处理后的。

JavaPairRDD<Object, BSONObject> updates = documents.mapValues(new MongoDBMapFunction());

mapFunction很简单，替换所有的name值为Spark，当然也可以做些统计的DEMO，后面的文章会再介绍更复杂的DEMO，敬请关注。

public class MongoDBMapFunction implements Function<BSONObject, BSONObject> {

    public BSONObject call(BSONObject bsonObject) throws Exception {
        bsonObject.put("name", "spark");
        return bsonObject;
    }
}

最后一步，数据的输出，MongoDB即是输入源又是输出源，所以第一个hdfs路径参数实际是无效的，但不可以是null，后面的类型描述了RDD的key，value类型，要跟updates一致，最后的config内容已经在程序最开始设置过了。

updates.saveAsNewAPIHadoopFile(
      "file://this-is-completely-unused",
      Object.class,
      BSONObject.class,
      MongoOutputFormat.class,
      config
);

额外说说明一下，Spark在动作是lazy的，整个代码流程下来，只有当程序执行到saveAsNewAPIHadoopFile时，才会触发数据拉取和计算等动作。

最后一步，构建Jar包，使用assembly的方式去构建，避免ClassNotFound的尴尬：

mvn assembly:assembly

上传JAR包并执行

剩下的操作都不需要写代码了，只需要操作控制台即可。几个步骤：上传JAR包->创建作业->创建执行计划->执行，我们来实际操作下。

再次登陆到OSS控制台，把刚才Jar包上传到OSS上，后面会用到。再回到EMR控制台上的作业栏里创建一个作业，需要指定一些参数，只名Job Class，然后点击下面的按钮添加OSS路径，内容是就是刚才上传的Jar包地址。值得注意的是，这里用的是ossref前缀，遇到这样的前缀EMR服务会自动的从OSS拉取Jar下来，否则原生的Spark是不识别的。最后应用参数应该是如下样子：

--master yarn-client --class com.aliyun.apsaradb.mongodb.Main ossref://sparkbucket/jar/spark-test-1.0-SNAPSHOT-jar-with-dependencies.jar

接下来是创建执行计划了，根据提示，在执行计划栏里进行创建，会提示采用的集群，作业集合，调度方式，这个DEMO采用的手动方式调度。

最后激动的时刻来临了，在执行计划栏里点击立即执行，运行过程和结束后都可以通过浏览器在网页上查看运行日志，非常方便。等待几十秒后，任务成功。

我们在回到DMS上查看数据集合，会发现已经多出了ouput集合，并且内容都为

{ "name": "spark"}

至此，Spark与MongoDB的Hello World风格教程结束，各位可以发挥无限的想象力，玩的开心！

参考连接：

时间： 2024-08-07 01:36:00

阿里云MongoDB与EMR的HelloWorld

EMR服务申请和创建

准备工作

申请EMR公测资格

创建EMR集群

购买阿里云云数据库MongoDB

检查网络连通性

Spark任务编写

Jar包依赖

Job编写

上传JAR包并执行

参考连接：

阿里云MongoDB与EMR的HelloWorld的相关文章

基于MongoDB构建阿里云MongoDB云数据库服务

阿里云MongoDB云服务构建

阿里云MongoDB Sharding备份和恢复服务深度解密

图解故障服务器下线：关于阿里云MongoDB高可用的探秘

阿里云CentOS7安装Mongodb教程_MongoDB

阿里云数据库，破解大型网站架构设计中的数据存储难题

116期：阿里云发布30+新产品

阿里云参与Redis中国用户组背后：不仅仅提供服务，也愿意分享技术

梨视频：基于阿里云E-MapReduce搭建视频推荐系统的实践