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

1
git clone https://github.com/mongodb-developer/data-lake-tutorial.git

1
from pymongo import MongoClient
2
from datetime import datetime
3
4
client = MongoClient('mongodb+srv://<username>:<password>@m0.lbtrerw.mongodb.net/')
5
…

1
pip3 install -r requirements.txt
2
python3 insert_data.py

1
{
2
  "databases": [
3
    {
4
      "name": "test",
5
      "collections": [
6
        {
7
          "name": "orders",
8
          "dataSources": [
9
            {
10
              "storeName": "M0",
11
              "database": "test",
12
              "collection": "orders"
13
            },
14
            {
15
              "storeName": "cold-data-mongodb",
16
              "path": "/*"
17
            }
18
          ]
19
        }
20
      ],
21
      "views": []
22
    }
23
  ],
24
  "stores": [
25
    {
26
      "name": "M0",
27
      "provider": "atlas",
28
      "clusterName": "M0",
29
      "projectId": "<your-project-id>"
30
    },
31
    {
32
      "name": "cold-data-mongodb",
33
      "provider": "s3",
34
      "bucket": "cold-data-mongodb",
35
      "prefix": "",
36
      "delimiter": "/"
37
    }
38
  ]
39
}

1
from datetime import datetime
2
3
from pymongo import MongoClient
4
5
client = MongoClient('<YOUR_FEDERATED_INSTANCE_URI>')
6
db = client.get_database('test')
7
coll = db.get_collection('orders')
8
9
start_date = datetime(2020, 5, 1)  # May 1st
10
end_date = datetime(2020, 6, 1)  # 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 archive.py

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)  # May 1st
10
end_date = datetime(2020, 6, 1)  # 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

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_FEDERATED_INSTANCE_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
python3 federated_queries.py

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}