POC pipeline to update a region of a zarr store, running from AWS Lambda and integrated with Prefect
Notes: the Zarr stores used are subsets of the actual big ones that we have for experimental purposes.
The subsets contain aggregated data of multiple NetCDF files (in smaller amounts) into Zarr stores, which is good enough to demonstrate how a Zarr store's regions can be updated.
📊 Pipeline run's effects (appending new data
, overwriting with new data
, overwriting with empty data
) on the Zarr stores (SST
, Argo
) are demonstrated in a separated Jupyter notebook.
https://aws.amazon.com/premiumsupport/knowledge-center/lambda-sqs-scaling
The maximum batch size for a standard Amazon SQS queue is 10,000 records. For FIFO queues, the maximum batch size is 10 records.
SQS FIFO was implemented but limited to 10 messages in a single batch, uploading lots of files will get the order we don't expect, see more: https://aws.amazon.com/blogs/compute/new-for-aws-lambda-sqs-fifo-as-an-event-source/
SQS standard queue doesn't guarantee order:
Standard queues provide best-effort ordering
-
Raw bucket is
update-zarr-region-raw
, when a NetCDF file being removed or uploaded to this bucket, it'll invokethe Lambda functionthe pipeline, which is a Lambda functionUpdateZarrRegionS3Watch
that sends event details as a message to a SQS FIFO Queue.UpdateZarrRegion
subscribes to the SQS FIFO queue and invoked when receiving new messages.to process the file. -
The pipeline powered by Prefect, is deployed by an image hosted on an AWS ECR private repository.
-
New GitHub code commits will trigger Github Actions to automatically build and push the latest image to ECR repository.
-
Untagged images (older images) will be cleaned up by an ECR Lifecycle policy settings.
-
An AWS CodePipeline entity watches the ECR repository changes and invokes a generic Lambda function (
UpdateLambdaFunctionCode
) that refreshes the pipeline with latest ECR image to make sure upcoming invocations will run with latest code changes.
Batch of multiple files uploaded/removed will trigger multiple Lambda processes running in parallel as Lambda is a distributed system.
We need to use zarr.sync.ProcessSynchronizer(path)
to HOPEFULLY ensure write consistency and avoid data corruptions.
-
Zarr Docs:
https://zarr.readthedocs.io/en/stable/api/sync.html
Path to a directory on a file system that is shared by all processes. N.B., this should be a different path to where you store the array.
Mutilple Lambda invocations need to access a shared directory to update and look for Zarr sync files while writing actual data to S3 Zarr store.
EFS
provides this feature out-of-the-box, NO "hackaround" here as EFS is specically designed for this purpose. -
AWS Docs:
Amazon EFS is a fully managed, elastic, shared file system designed to be consumed by other AWS services, such as Lambda. With the release of Amazon EFS for Lambda, you can now easily share data across function invocations.
Multiple compute instances, including ... AWS Lambda, can access an Amazon EFS file system at the same time, providing a common data source for workloads
Go to toolkits/handlers
and create a new package for new data collection. 3 files required:
__init__py
with empty content.config.json
where you store information of the data collection essential information such as Zarr store path, constants, dimensions, chunks etc.handler.py
takes abstract classDataset
as a parameter, you'll need to provide specific implementations depends on the data collections to the abstract methods.
Multi-dimensions data of different data collections may have different implementations to locate a Zarr store's region, generate empty dataset, and so on. For example, you really CANNOT make the implementations for ARGO to be generic and applicable for other collections such as SST, they have completely different structures.
Then update handler_masks.json
with new data collection handler package information including the Regex
pattern of the S3 object_key
.
And finally, import new handler package to utils.py
, methods in this file are wrapped with Prefect
task decorator and are generic.
-
Update Zarr store regions is doable when ingested NetCDF files in
raw
bucket removed or updated, as long as there are no other processs writing data to the Zarr stores, evenProcessSynchronizer()
applied on a distributed system like Lambda. -
How to do it in large scale is a topic to discuss further, processing multiple files in order 1-by-1 in the same process is a safe approach.
-
Updating a region requires to process one file only per transaction, it helps identify which regions to be updated correctly. E.g Zarr store with 5 regions: [1, 2, 3, 4, 5], imagine only [2,5] to be updated, concatinated datasets of [2,5] cannot help identifying appropriate region, only if we want to overwrite all regions sitting next to each other, start from [2] and stop at [5]: [2,3,4,5], but this is not ideal, what if the Zarr stores has 5000 regions and only [100] and [4001] need to be updated? - Stick with processing 1 file per transaction is a safe approach.
-
Appending new files coming in a batch can be done with multiple files per transaction. Just because we can, we should?
-
Event-driven with Lambda that triggering by every single file to start a process writing to Zarr will cause messy results even the processes have shared access to Synchronizer files on EFS.
-
Consider aggregating data pipelines to Zarr to be on scheduled flows OR some hackaround methods to let the pipelines wait until receiving enough files, eg. wait until receiving 100 files then firing the flows, while there's a processing writing to Zarr, block other processes. If do this way, investigate how to check if there is an existing process writing to Zarr store?