# SLZ DEMO 2020-05-12 # Plan: 1. Terminology 1. DSP 1. UOW statuses and flow 1. Content Status record statuses and flow 1. Stack 1. Buckets 1. Downloader 1. Main SF + Job Manager + Reprocessing 1. MetaFlow SF ## Treminology Single Landing Zone (SLZ) - S3-based data lake solution. ETL - Process that uses downloaded files. Digital Service Provider (DSP) - source of the data. Licensor - Sony, The Orchard, ... Unit of Work (UOW) - artificial unit we use for tracking the status of report files. Usually we treat a combination of DSP, report type, report version, licensor and date as a unique identifier for set of files. `unit_of_work` table keeps Unit of Work records. Context - this is something DSP specific. Context is bound to the real file we have in data lake in the end. In SLZ database a record in a `content_status` table reflects the state of Context. For Spotify we consider that country is a context (report for DE, US, FR, ...), for Apple - this is a combination of region + vendor_id (1682::80029727, 1802::80030238, ...) Context can be `required`. This means that as soon as all `required` contexts are downloaded unit of work will be marked as `MIN_COMPLETE` and ETL process can be started. When all contexts are downloaded its unit of work becomes `COMPLETE` and ETL can produce the final version of report. ## DSP List of DSP we support in SLZ: - HTTP API based - Apple, Spotify, Youtube - S3 based - Amazon, Vevo, Tiktok, Linkfire, Sme_Max, Youtube - SQL Database - GRAS, Apollo ## UOW Statuses and flow Activity Status: IN_PROGRESS NOT_IN_PROGRESS Content Status: ACTIVE - this UOW is not finished yet MIN_COMPLETE - required contexts are downloaded and we can start process them in ETL COMPLETE - all contexts are downloaded CANCELLED - for some reasons such UOW was cancelled during the process ON_HOLD - something goes wrong and we cannot fix the process automatically Ideally: ACTIVE -> MIN_COMPLETE -> COMPLETE Get list of unit of works: select * from unit_of_work limit 10 ## Content Status and flow Content status: ON_HOLD - file was not processed properly, because of size difference ACTIVE - file created and probably in progress FAILED - some error appeared MISSING - file is not ready to be downloaded CANCELLED - process was cancelled COMPLETE - file downloaded and passed set of validation rules Ideally: MISSING -> ACTIVE -> COMPLETE Errors during the execution ACTIVE -> FAILED File downloaded correctly, but failed on size validation ACTIVE -> ON_HOLD File not found ACTIVE-> MISSING Reprocessing is started, current execution cancelled ACTIVE -> CANCELLED # Implementation details ## Stack - Python 3.10 - Docker - AWS S3 - AWS Secret Manager - AWS Aurora Postgres - AWS Lambda - AWS Step Function - AWS Batch - AWS ECR - AWS Dynamodb - dead CI/CD: - Jenkins - Octopu ## Repositories Python code: [https://github.com/filtr/delphi-slz](https://github.com/filtr/delphi-slz) - SLZ main repository [https://github.com/filtr/delphi-util](https://github.com/filtr/delphi-util) - utils / libraries [https://github.com/filtr/delphi-config](https://github.com/filtr/delphi-config) - configs Set of configs that we use in SLZ: https://github.com/filtr/delphi-config/blob/master/configs/prod/dsp_config.json - dsp_config https://github.com/filtr/delphi-config/blob/master/configs/prod/dsp-specific-settings.json - dsp specific settings https://github.com/filtr/delphi-config/blob/master/configs/prod/dsp-complete-criteria.json - dsp complete criteria Configs on S3: https://s3.console.aws.amazon.com/s3/buckets/stage-delphi-configs/?region=us-east-1 Shemas: https://github.com/filtr/data-dictionary Examples: Spotify: https://github.com/filtr/data-dictionary/blob/master/schemas/spotify/streams/v2/2017-09-04.json Apple: https://github.com/filtr/data-dictionary/blob/master/schemas/apple/amArtistDemographics/v1_1/2018-12-13.json ## Buckets Corrupted files: https://s3.console.aws.amazon.com/s3/buckets/stage-delphi-sme-data-corrupted/?region=us-east-1 Quarantine archive: https://s3.console.aws.amazon.com/s3/buckets/stage-delphi-quarantine-sme-data-archive/?region=us-east-1 Quarantine decompressed: https://s3.console.aws.amazon.com/s3/buckets/stage-delphi-quarantine-sme-data-decompressed/?region=us-east-1 Public archive: https://s3.console.aws.amazon.com/s3/buckets/stage-sme-data-archive/?region=us-east-1 Public decompressed: https://s3.console.aws.amazon.com/s3/buckets/stage-sme-data-decompressed/?region=us-east-1&tab=overview ## Downloader Python application we launch using AWS Batch. This application can be scaled to fully utilize CPU of EC2 instance, but cannot be scaled across the multiple nodes at least for now. Input: - list of application configs - list of contexts to process - names of buckets Output: - list of contexts with statuses of success/failure Flow: 1. Send context to separate system process 1. Update content_status record 1. Open stream and start downloading 1. Compressed stream will be decompressed (on a fly if possible) and raw stream will be compressed. 1. Upload stream into `archive-quarantine` and `decompressed-quarantine` buckets 1. Check size of reports - compare its size with size of a file on S3 1. Validate first N records with schemas (`data-dictionaries` repository) 1. Validate size of reports, compare it to the average size of this report for the last M weeks (Historical validation) 1. Move files into `archive` and `decompressed` buckets 1. Mark content_status as `COMPLETE` 1. Send a notification to `SQS` queue that new file has landed on S3. If download process fails on Historical validation - change completeness status to `ON_HOLD`. Otherwise - `FAILED`. File goes to `corrupted` bucket. ## Main SF + Job Manager + Reprocessing Job Manager lambda starts execution of Step Functions Regular event ``` { "id": "cdc73f9d-aea9-11e3-9d5a-835b769c0d9c", "detail-type": "Scheduled Event", "source": "aws.events", "account": "{{account-id}}", "time": "2020-05-10T10:00:50Z", "region": "us-east-1", "resources": [ "arn:aws:events:us-east-1:123456789012:rule/ExampleRule" ], "detail": {} } ``` Reprocessing event ``` { "id": "cdc73f9d-aea9-11e3-9d5a-835b769c0d9c", "detail-type": "Scheduled Event", "source": "aws.events", "account": "{{account-id}}", "time": "2020-05-04T21:40:50Z", "region": "us-east-1", "resources": [ "arn:aws:events:us-east-1:123456789012:rule/ExampleRule" ], "reprocessing": [ { "uow_id": "apollo-20200508-sme-spotify_charts-v1", "priority": 5 } ], "detail": {} } ``` Job manager lambda https://console.aws.amazon.com/lambda/home?region=us-east-1#/functions/stage-delphi-slz-job-manager-stage?tab=configuration ## Main Step Function https://console.aws.amazon.com/states/home?region=us-east-1#/executions/details/arn:aws:states:us-east-1:475275892927:execution:stage-delphi-slzMainFlow:24458_2020-05-12T12.23.53 ## Meta Flow Step Function https://console.aws.amazon.com/states/home?region=us-east-1#/executions/details/arn:aws:states:us-east-1:475275892927:execution:stage-delphi-slzMetaFlow:22617_2020-05-12T12_39_12_162331