MongoDB on Spark的参数优化

分手后的思念是犯贱 2022-05-03 02:58 144阅读 0赞

## MongoDB on Spark的参数优化 ##

### mongodb on spark的参数： ###

mongodb on spark的参数基本分为输入参数和输出参数。

#### 1. mongodb on spark输入的参数： ####

*  spark.mongodb.input.uri：mongo的uri，
    
    如：mongodb://host:port/
    
    或者直接指定数据库和集合：mongodb://host:port/database.collection，相当于使用了uri，database，collection参数
 *  spark.mongodb.input.database：mongo的数据库指定
 *  spark.mongodb.input.collection：mongo的集合
 *  **spark.mongodb.input.partitioner**：用于对数据进行分区的分区程序的类名
    
    默认使用：MongoDefaultPartitioner，其他值有：
    
     *  MongoSamplePartitioner：使用集合的平均文档大小和随机抽样来确定集合的合适分区。
     *  MongoShardedPartitioner：根据数据块对集合进行分区。需要对config数据库的读访问权限。
     *  MongoSplitVectorPartitioner：使用splitVector独立命令或主数据库上的命令来确定数据库的分区。需要特权才能运行splitVector命令
     *  MongoPaginateByCountPartitioner：创建特定数量的分区。需要查询每个分区。
     *  MongoPaginateBySizePartitioner：根据数据大小创建分区。需要查询每个分区。
    
    每个Partitioner类都有option参数，具体参阅[文档][Link 1]

#### 2. mongodb on spark输出的参数 ####

*  spark.mongodb.output.uri：和输入参数一样
 *  spark.mongodb.output.database：和输入参数一样
 *  spark.mongodb.output.collection：和输入参数一样
 *  spark.mongodb.output.replaceDocument：保存包含\_id字段的数据集时替换整个文档。如果为false，它将仅更新文档中与数据集中的字段匹配的字段。默认true。
 *  maxBatchSize：保存数据时批量操作的最大批量大小。默认512M。

--------------------

#### 3. mongodb on spark的参数添加 ####

mongodb on spark的参数添加有2种方式，可以通过spark的conf添加，也可以通过option添加参数的方式进行添加。

3.1 通过config的方式进行添加

val spark = SparkSession.builder()
          .master("local")
          .appName("MongoSparkConnectorIntro")
          .config("spark.mongodb.input.uri", "mongodb://127.0.0.1/test.myCollection")
          .config("spark.mongodb.output.uri", "mongodb://127.0.0.1/test.myCollection")
          .getOrCreate()

这里也可以通过config添加其他参数。

3.2 通过option的方式进行添加

# 输入参数添加
    
    spark.read.format("com.mongodb.spark.sql.DefaultSource") \
        .option("spark.mongodb.input.partitioner", "MongoPaginateByCountPartitioner") \
        .option("spark.mongodb.input.partitionerOptions.partitionKey", 'service_time') \
        .option("spark.mongodb.input.partitionerOptions.numberOfPartitions", "10") \
        .schema(collect_schema) \
        .load()
    
    
    
    # 输出参数添加
    
    df.write.format('com.mongodb.spark.sql.DefaultSource')
        .option("spark.mongodb.output.uri", "mongodb://127.0.0.1/test.myCollection") \
        .mode("append")\
        .save()

--------------------

参考文档：  
[MongoDB on Spark的参数优化][MongoDB on Spark]

[Link 1]: https://docs.mongodb.com/spark-connector/current/configuration/#partitioner-conf
[MongoDB on Spark]: https://docs.mongodb.com/spark-connector/current/configuration/