Spark 连接 Mongodb 批量读取数据

Spark 连接 mongodb ,并多次切换集合

方案一：通过 JavaSparkContext 连接 mongodb ，利用 MongoSpark.load() 方法获取集合数据

测试 demo 如下：

import com.mongodb.spark.MongoSpark;
import com.mongodb.spark.config.ReadConfig;
import com.mongodb.spark.rdd.api.java.JavaMongoRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FilterFunction;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.bson.Document;

import java.util.ArrayList;
import java.util.HashMap;

public class SparkReadMongodbs {
    public static void main(String[] args) {
        String mongoUrl="mongodb://root:root123456@192.168.1.124:27017,192.168.1.123:27017,192.168.1.125:27017/";
        String database="lhiot";
        String dbCollection="0762a06a97b3628bd00037e6f66c7d16";
        String port = "27017";

        SparkSession.Builder builder =SparkSession.builder().master("local[*]").appName("SparkCalculateRecommend")
                .config("spark.mongodb.input.uri", mongoUrl+database+"."+dbCollection+"?authSource=admin")
                .config("spark.executor.memory", "512mb");

        SparkSession spark = builder.getOrCreate();

        JavaSparkContext jsc = new JavaSparkContext(spark.sparkContext());
        //使用Spark连接器载入sparkContext，获取RDD对象
        JavaMongoRDD<Document> c1 = MongoSpark.load(jsc);

        ArrayList<String> collections = new ArrayList<>();
        collections.add("00dfaed143dcbb02ae21aaec492d369d");
        collections.add("020a91e9c60fab73d244ba797c485e47");
        collections.add("02a70e55a7ff1a4ebb4dbbeb3e28c137");
        collections.add("0588dee7e8fdde3d95ba250affeab843");
        collections.add("0762a06a97b3628bd00037e6f66c7d16");
        collections.add("0914e6088a799c8cee11df25e11e2534");
        collections.add("0f768fc73fed9752fd87f432e9d77ba6");
        collections.add("1336a41b0bd13e1ca6a86905b9c6fd9d");
        collections.add("1ea1b22693d1bdb592853ec59c4d1fe3");

        HashMap<String, String> readOverrides = new HashMap<>();

        for (String collection : collections) {
            readOverrides.put("collection", collection);
            //读取数据库对应集合数据
            ReadConfig readConfig = ReadConfig.create(jsc).withOptions(readOverrides);
            //获取该设备集合数据
            JavaMongoRDD<Document> c2 = MongoSpark.load(jsc,readConfig);
            c2.toDF()
                    .select("_id.oid", "deviceCode", "funCode", "deptId", "deptName", "mountId", "mountName", "deviceId",
                            "pointId", "pointName", "pointOrderNum", "value", "pointDisplayName", "unit", "originTime", "createTime")
                    .withColumnRenamed("oid", "id")
                    .filter(new FilterFunction<Row>() {
                        @Override
                        public boolean call(Row value) throws Exception {
                            String originTime = value.getAs("originTime").toString();
                            return originTime.compareTo("2022-01-22 00:00:00")>=0 && originTime.compareTo("2022-01-22 23:59:59")<=0;
                        }
                    })
                    .show();
        }
		jsc.stop();
        spark.stop();
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66

该方法，在切换集合时，会产生大量的新增连接，程序结束，所有连接会断开。
但是如果业务需要从大量的集合中读取数据，这个方式就不太合适了，维护大量的连接，spark会消耗大量的内存，同事mongo端也会有很大压力，甚至会导致数据库服务的挂掉。

方案二：通过 JavaSparkContext 连接 mongodb ，利用 sqlContext.read().load() 方法获取集合数据

测试 demo 如下：

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SQLContext;
import org.apache.spark.sql.SparkSession;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.Map;

public class sparkReadMongodbWithoutCol {
    public static void main(String[] args) {
        String mongoUrl="mongodb://root:root123456@192.168.1.124:27017,192.168.1.123:27017,192.168.1.125:27017/";
        String database="lhiot";
        String dbCollection="0762a06a97b3628bd00037e6f66c7d16";
        String port = "27017";

        //将options的配置信息存储到一个map里
        Map<String, String> map = new HashMap<String, String>();
//        map.put("uri",mongoUrl);
        map.put("database", database);
//        map.put("collection", dbCollection);

        //连接mongodb服务器
        SparkConf sc = new SparkConf().setMaster("local").setAppName("SparkConnectMongo")
                .set("spark.app.id", "MongoSparkConnectorTour")
                .set("spark.mongodb.input.uri", mongoUrl + "?authSource=admin")
                .set("spark.testing.memory","471859200");
        JavaSparkContext jsc = new JavaSparkContext(sc);
        SQLContext sqlContext = new SQLContext(jsc);

        ArrayList<String> collections = new ArrayList<>();
        collections.add("0762a06a97b3628bd00037e6f66c7d16");
        collections.add("00dfaed143dcbb02ae21aaec492d369d");
        collections.add("020a91e9c60fab73d244ba797c485e47");
        collections.add("02a70e55a7ff1a4ebb4dbbeb3e28c137");
        collections.add("0588dee7e8fdde3d95ba250affeab843");
        collections.add("0762a06a97b3628bd00037e6f66c7d16");
        collections.add("0914e6088a799c8cee11df25e11e2534");
        collections.add("0f768fc73fed9752fd87f432e9d77ba6");
        collections.add("1336a41b0bd13e1ca6a86905b9c6fd9d");
        collections.add("1ea1b22693d1bdb592853ec59c4d1fe3");

        for (String collection : collections) {
            map.put("collection", collection);
            //读取数据库对应集合数据
            Dataset<Row> res = sqlContext.read().format("com.mongodb.spark.sql").options(map).load();
            res.registerTempTable("table");
            sqlContext.sql("select * from table").show();
        }

        jsc.stop();
    }

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56

该方案再切换不同集合时，不会产生大量的连接，整个程序只会在mongo端产生2个连接，程序结束，2个连接也会自动断开。
该方案就比较适合需要同时读取大量集合数据的需求场景。

以上只是我的简单测试方案，理解较为浅显，欢迎大佬留言交流，谢谢鉴赏。