MongoDB Atlas Data Lake Tutorial: Federated Queries and $out to AWS S3

1
{
2
  "databases": [
3
    {
4
      "name": "test",
5
      "collections": [
6
        {
7
          "name": "orders",
8
          "dataSources": [
9
            {
10
              "path": "/{min(created) isodate}-{max(created) isodate}*",
11
              "storeName": "cold-data-mongodb"
12
            },
13
            {
14
              "collection": "orders",
15
              "database": "test",
16
              "storeName": "BigCluster"
17
            }
18
          ]
19
        }
20
      ],
21
      "views": []
22
    }
23
  ],
24
  "stores": [
25
    {
26
      "provider": "s3",
27
      "bucket": "cold-data-mongodb",
28
      "delimiter": "/",
29
      "includeTags": false,
30
      "name": "cold-data-mongodb",
31
      "region": "eu-west-1"
32
    },
33
    {
34
      "provider": "atlas",
35
      "clusterName": "BigCluster",
36
      "name": "BigCluster",
37
      "projectId": "5e78e83fc61ce37535921257"
38
    }
39
  ]
40
}

1
from datetime import datetime
2
3
from pymongo import MongoClient
4
5
client = MongoClient('<YOUR_ATLAS_DATA_LAKE_URI>')
6
db = client.get_database("test")
7
coll = db.get_collection("orders")
8
9
start_date = datetime(2020, 5, 1, 0, 0, 0)  # May 1st
10
end_date = datetime(2020, 6, 1, 0, 0, 0)  # June 1st
11
12
pipeline = [
13
    {
14
        '$match': {
15
            'created': {
16
                '$gte': start_date,
17
                '$lt': end_date
18
            }
19
        }
20
    },
21
    {
22
        '$out': {
23
            's3': {
24
                'bucket': 'cold-data-mongodb',
25
                'region': 'eu-west-1',
26
                'filename': start_date.isoformat('T', 'milliseconds') + 'Z-' + end_date.isoformat('T', 'milliseconds') + 'Z',
27
                'format': {'name': 'json', 'maxFileSize': "200MiB"}
28
            }
29
        }
30
    }
31
]
32
33
coll.aggregate(pipeline)
34
print('Archive created!')

1
$ python3 --version
2
Python 3.8.3
3
$ pip3 install pymongo dnspython
4
Requirement already satisfied: pymongo in /home/polux/.local/lib/python3.8/site-packages (3.10.1)
5
Requirement already satisfied: dnspython in /home/polux/.local/lib/python3.8/site-packages (1.16.0)
6
$ python3 archive.py
7
Archive created!

1
from datetime import datetime
2
3
from pymongo import MongoClient
4
5
client = MongoClient('<YOUR-ATLAS-URI>')
6
db = client.get_database("test")
7
coll = db.get_collection("orders")
8
9
start_date = datetime(2020, 5, 1, 0, 0, 0)  # May 1st
10
end_date = datetime(2020, 6, 1, 0, 0, 0)  # June 1st
11
query = {
12
    'created': {
13
        '$gte': start_date,
14
        '$lt': end_date
15
    }
16
}
17
18
result = coll.delete_many(query)
19
print('Deleted', result.deleted_count, 'orders.')

1
$ python3 remove.py
2
Deleted 2 orders.

1
{"_id":{"$numberDouble":"1.0"},"created":{"$date":{"$numberLong":"1590796800000"}},"items":[{"$numberDouble":"1.0"},{"$numberDouble":"3.0"}],"price":{"$numberDouble":"20.0"}}
2
{"_id":{"$numberDouble":"2.0"},"created":{"$date":{"$numberLong":"1590883200000"}},"items":[{"$numberDouble":"2.0"},{"$numberDouble":"3.0"}],"price":{"$numberDouble":"25.0"}}

1
from pymongo import MongoClient
2
3
client = MongoClient('<YOUR_ATLAS_DATA_LAKE_URI>')
4
db = client.get_database("test")
5
coll = db.get_collection("orders")
6
7
print("All the docs from S3 + Atlas:")
8
docs = coll.find()
9
for d in docs:
10
    print(d)
11
12
pipeline = [
13
    {
14
        '$group': {
15
            '_id': None,
16
            'total_price': {
17
                '$sum': '$price'
18
            }
19
        }
20
    }, {
21
        '$project': {
22
            '_id': 0
23
        }
24
    }
25
]
26
27
print('\nI can also run an aggregation.')
28
print(coll.aggregate(pipeline).next())

1
All the docs from S3 + Atlas:
2
{'_id': 1.0, 'created': datetime.datetime(2020, 5, 30, 0, 0), 'items': [1.0, 3.0], 'price': 20.0}
3
{'_id': 2.0, 'created': datetime.datetime(2020, 5, 31, 0, 0), 'items': [2.0, 3.0], 'price': 25.0}
4
{'_id': 3.0, 'created': datetime.datetime(2020, 6, 1, 0, 0), 'items': [1.0, 3.0], 'price': 20.0}
5
{'_id': 4.0, 'created': datetime.datetime(2020, 6, 2, 0, 0), 'items': [1.0, 2.0], 'price': 15.0}
6
7
I can also run an aggregation:
8
{'total_price': 80.0}