ApsaraDB for MongoDB:View the progress of adding or removing shards

Methods available for versions earlier than MongoDB 6.0

Before you evaluate a task, you must obtain the operational results of the balancer and the chunk information of a sharded collection to be migrated. The operational results of the balancer include statistics on the number of chunks that are migrated and chunks that fail to be migrated.

You can use one of the following methods to obtain the preceding information:

• Method 1: View the output returned for the sh.status() command

  In the output returned for the sh.status() command, you must focus on the recent operational results of the balancer and the chunk information of a sharded table to be migrated. The following example shows the recent operational results of the balancer:

  ...
    balancer:
          Collections with active migrations: 
                  <db>.<collection> started at Wed Sep 27 2023 10:25:21 GMT+0800 (CST)
          Failed balancer rounds in last 5 attempts:  0
          Migration Results for the last 24 hours: 
                  300 : Success
    databases:
  ...

  The following example shows the chunk information of a sharded collection to be migrated:

  ...
  databases:
  ...
      {  "_id" : "<db>",  "primary" : "d-xxxxxxxxxxxxxxx3",  "partitioned" : true,  "version" : {  "uuid" : UUID("3409a337-c370-4425-ad72-8b8c6b0abd52"),  "lastMod" : 1 } }
            <db>.<collection>
                          shard key: { "<shard_key>" : "hashed" }
                          unique: false
                          balancing: true
                          chunks:
                                  d-xxxxxxxxxxxxxxx1      13630
                                  d-xxxxxxxxxxxxxxx2      13629
                                  d-xxxxxxxxxxxxxxx3      13652
                                  d-xxxxxxxxxxxxxxx4      13630
                                  d-xxxxxxxxxxxxxxx5      3719
                          too many chunks to print, use verbose if you want to force print
  ...

  In the preceding example, d-xxxxxxxxxxxxxxx5 indicates a shard that is added. You can also run the getShardDistribution command on the database where a partitioned table is located to obtain information about the distribution of data chunks. Sample code:

  use <db>
  db.<collection>.getShardDistribution()

• Method 2: Directly read related information from the config database

  Query the statistics on chunks and then aggregate them by shard. Sample code:

  db.getSiblingDB("config").chunks.aggregate([{$group: {_id: "$shard", count: {$sum: 1}}}])

  Query the chunks on a specified shard and then aggregate them by namespace. Sample code:

  db.getSiblingDB("config").chunks.aggregate([{$match: {shard: "d-xxxxxxxxxxxxxx"}},{$group: {_id: "$ns", count: {$sum: 1}}}])

  Query the number of chunks that are migrated to a specified shard within the previous day. Sample code:

  // Specify the details.to field as the shard whose chunks you want to migrate.
  // Specify the time field as a time range of the ISODate type.
  db.getSiblingDB("config").changelog.find({"what" : "moveChunk.commit", "details.to" : "d-xxxxxxxxxxxxx","time" : {"$gte": ISODate("2023-09-26T00:00:00")}}).count()

Methods available for versions later than MongoDB 6.0

Before you evaluate a task, you must focus on the data distribution among shards. You can still use the output returned by the sh.status() command as a reference. However, after you obtain the output, you must pay more attention to the data distribution on each shard instead of the uneven number of chunks on each shard.

• Obtain the data distribution and uniformity of a sharded collection.

  We recommend that you run the getShardDistribution command to obtain the data distribution on a sharded collection and focus on the data uniformity, as shown in the following figure.

  You can also run the following command to obtain more detailed distribution, including the number and total size of documents and that of orphaned documents:

  db.getSiblingDB("admin").aggregate( [{ $shardedDataDistribution: { } },{ $match: { "ns": "<db>.<collection>" } }] ).pretty()

  The following example shows an output for the preceding command. The output indicates that data is significantly uneven among shards.

  {
          "ns" : "<db>.<collection>",
          "shards" : [
                  {
                          "shardName" : "d-xxxxxxxxxxxxxxx1",
                          "numOrphanedDocs" : 0,
                          "numOwnedDocuments" : 504298920,
                          "ownedSizeBytes" : NumberLong("833101815840"),
                          "orphanedSizeBytes" : 0
                  },
                  {
                          "shardName" : "d-xxxxxxxxxxxxxxx2",
                          "numOrphanedDocs" : 0,
                          "numOwnedDocuments" : 250283901,
                          "ownedSizeBytes" : NumberLong("409714745937"),
                          "orphanedSizeBytes" : 0
                  },
                  {
                          "shardName" : "d-xxxxxxxxxxxxxxx3",
                          "numOrphanedDocs" : 0,
                          "numOwnedDocuments" : 109098088,
                          "ownedSizeBytes" : NumberLong("178157177704"),
                          "orphanedSizeBytes" : 0
                  },
                  {
                          "shardName" : "d-xxxxxxxxxxxxxxx4",
                          "numOrphanedDocs" : 0,
                          "numOwnedDocuments" : 382018055,
                          "ownedSizeBytes" : NumberLong("630329790750"),
                          "orphanedSizeBytes" : 0
                  }
            ]
  }

  The output also indicates the following items: 1. The instance on which you run the preceding command contains four shards. 2. The data volume of a sharded collection named <db>.<collection> in the instance is about 2051303530231 bytes (833101815840 + 409714745937 + 178157177704 + 630329790750 ), which is equivalent to 1910.4 GB.

• View the amount of data that was successfully migrated within the previous day.

  Run the following command:

  pipeline = [
      { '$match': {
          'what': 'moveChunk.commit',
      } },
      { '$group': {
          '_id': {
              'date': { '$dateToString': { 'format': '%Y-%m-%d', 'date': '$time'} },
           },
          'chunks_moved': { '$sum': 1 },
          'docs_moved': { '$sum': '$details.counts.cloned' },
          'bytes_moved': { '$sum': '$details.counts.clonedBytes' },
      } }, 
      { '$sort': { '_id.date': -1} },
  ]
  db.getSiblingDB("config").changelog.aggregate(pipeline)

  The following figure shows the output returned for the preceding command.

  The output indicates that 183117965820 bytes are successfully migrated on 2024-09-21, which is equivalent to 170.5 GB.

Methods available for versions earlier than MongoDB 6.0

After you obtain the number of chunks that are migrated from a sharded collection and the distribution of data in the current sharded collection, you can estimate the overall progress and expected completion time of the task.

Assume that you want to add shards. Before shards are added, the total number of chunks remains unchanged. In addition, the number of shards remains unchanged, which indicates that no tasks are initiated to add or remove shards. When shards are being added, the business workload remains constant, and parameters related to the balancer are set to their default values. The first example described in methods available for versions earlier than MongoDB 6.0 in Step 3 is used. The example indicates the following items:

• Five shards are added.

• 300 chunks are migrated within a time window of the balancer.

• 58,260 chunks are contained in a sharded collection named <db>.<collection>. The number of chunks is calculated based on the following formula: 13,630 + 13,629 + 13,652 + 13,630 + 3,719 = 58,260.

The following values can be calculated based on the preceding results:

• When chunks are evenly distributed across shards, the number of chunks in each shard is 11,652 (58,260/5 = 11,652).

• Based on the current migration speed, it takes another 26.4 days to complete the migration. The number of days is calculated based on the following formula: (11,652 - 3,719)/300 ≈ 26.4.

• The process of the task for adding shards is 32% (3,719/11,652 = 32%).

Methods available for versions later than MongoDB 6.0

Assume that only one collection in a sharded cluster instance needs to be balanced. The first example described in methods available for versions later than MongoDB 6.0 in Step 3 is used. The example indicates the following items:

• Four shards are added.

• 183117965820 bytes are successfully migrated on 2024-09-21, which is equivalent to 170.5 GB.

• The data volume of a sharded collection named <db>.<collection> is about 2051303530231 bytes (833101815840 + 409714745937 + 178157177704 + 630329790750), which is equivalent to 1910.4 GB.

The following values can be calculated based on the preceding results:

• When data balance is reached among shards, the data volume of each shard is about 512825882557.75 bytes, which is equivalent to 477.6 GB.

• In order to achieve data balance among shards, you must migrate a total of 875559682949 bytes among the shards, which is equivalent to 815.4 GB. Specific data migration from each shard:

  The amount of data to be migrated from Shard 1 is 320275933282.25 bytes (833101815840 - 512825882557.75).

  The amount of data to be migrated from Shard 2 is 103111136620.75 bytes (409714745937 - 512825882557.75).

  The amount of data to be migrated from Shard 3 is 334668704853.75 bytes (178157177704 - 512825882557.75).

  The amount of data to be migrated from Shard 4 is 117503908192.25 bytes (630329790750 - 512825882557.75).

  (320275933282.25 + 103111136620.75 + 334668704853.75 + 117503908192.25 = 875559682949)

• It takes about 4.8 days (815.4 GB/170.5 GB per day) to reach data balance at the current migration rate.