Data lake in S3 from MongoDB

1_Data lake in S3 from MongoDB

# Data lake in S3 from MongoDB

auctions.yaml

collection_name: catawiki_auctions
incremental: true
date_field: createdAt
start_date: 2020-03-01
end_date: 2020-08-10
hours_interval: 24

compress.py

# Store file as compressed JSON
with gzip.GzipFile(file_name, "w") as file_out:
    file_out.write(json_bytes)

date_range.py

for i in range(1, 10):
    print(i, end=", ")

#=> 1, 2, 3, 4, 5, 6, 7, 8, 9, 


for day in date_range("2021-01-01", "2021-01-10"):
    print(day, end=", ")

#=> 2021-01-01 00:00:00, 2021-01-02 00:00:00, 2021-01-03 00:00:00, 2021-01-04 00:00:00, 2021-01-05 00:00:00, 2021-01-06 00:00:00, 2021-01-07 00:00:00, 2021-01-08 00:00:00, 2021-01-09 00:00:00,

file_name.py

file_name = f"{COLLECTION_NAME}_{date_str_file}_{z_lower_limit}-{z_upper_limit}_{today_str}_batch{batch_number}.json.gz"

# => catawiki_auctions_20200310_0000-2400_20211010082324540093_batch0.json.gz

json_serialize.py

import json

json_str = json.dumps(records)

# => TypeError: Object of type ObjectId is not JSON serializable

policy.json

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": "s3:*",
      "Resource": "arn:aws:s3:::BUCKET-NAME-HERE/*"
    }
  ]
}

pymongo_serialize.py

from bson.json_util import dumps as mongo_dumps

# Dump and encode records
json_str = mongo_dumps(records)
json_bytes = json_str.encode("utf-8")

upload.py

# Start an S3 client
s3_client = boto3.client(
    "s3",
    aws_access_key_id=AWS_ACCESS_KEY_ID,
    aws_secret_access_key=AWS_SECRET_ACCESS_KEY,
)

# Upload file to S3 bucket
s3_client.upload_file(
    file_path,
    bucket_name,
    bucket_path,
)