Manage cloud-hosted data

Move data to and from cloud provider storage.

Learn how to use Prefect to move data to and from AWS, Azure, and GCP blob storage.

Prerequisites

• Prefect installed
• Authenticated with Prefect Cloud (or self-hosted Prefect server instance)
• A cloud provider account

Install relevant Prefect integration library

In the CLI, install the Prefect integration library for your cloud provider:

AWS

prefect-aws provides blocks for interacting with AWS services.

pip install -U prefect-aws

Azure

prefect-azure provides blocks for interacting with Azure services.

 pip install -U prefect-azure

GCP

prefect-gcp provides blocks for interacting with GCP services.

 pip install -U prefect-gcp

Register the block types

Register the new block types with Prefect Cloud (or with your self-hosted Prefect server instance):

AWS

prefect block register -m prefect_aws  

Azure

prefect block register -m prefect_azure 

GCP

prefect block register -m prefect_gcp

A confirmation message in the CLI shows that several block types were registered. The UI shows the new block types listed.

Create a storage bucket

Create a storage bucket in the cloud provider account. Ensure the bucket is publicly accessible, or create a user or service account with the appropriate permissions to fetch and write data to the bucket.

Create a credentials block

If the bucket is private, there are several options to authenticate:

• At deployment runtime, ensure the runtime environment is authenticated.
• Create a block with configuration details and reference it when creating the storage block.

If you saved credential details in a block, you can use a credentials block specific to the cloud provider or use a more generic secret block. You can create blocks through the UI or Python code.

The example below uses Python code to create a credentials block for your cloud provider.

Credentials safety

Don’t store credential values in public locations such as public git platform repositories. The examples below use environment variables to store credential values.

AWS

import os
from prefect_aws import AwsCredentials

my_aws_creds = AwsCredentials(
    aws_access_key_id="123abc",
    aws_secret_access_key=os.environ.get("MY_AWS_SECRET_ACCESS_KEY"),
)
my_aws_creds.save(name="my-aws-creds-block", overwrite=True)

Azure

import os
from prefect_azure import AzureBlobStorageCredentials

my_azure_creds = AzureBlobStorageCredentials(
    connection_string=os.environ.get("MY_AZURE_CONNECTION_STRING"),
)
my_azure_creds.save(name="my-azure-creds-block", overwrite=True)

GCP

We recommend specifying the service account key file contents as a string, rather than the path to the file, because that file might not be available in your production environments.

import os
from prefect_gcp import GCPCredentials

my_gcp_creds = GCPCredentials(
    service_account_info=os.environ.get("GCP_SERVICE_ACCOUNT_KEY_FILE_CONTENTS"), 
)
my_gcp_creds.save(name="my-gcp-creds-block", overwrite=True)

Run the code to create the block. You should see a message that the block was created.

Create a storage block

You can create a block for the chosen cloud provider using Python code or the UI. This example uses Python code.

AWS

Note that the S3Bucket block is not the same as the S3 block that ships with Prefect. The S3Bucket block used in this example is part of the prefect-aws library and provides additional functionality.

Next, reference the credentials block created above.

from prefect_aws import S3Bucket

s3bucket = S3Bucket.create(
    bucket="my-bucket-name",
    credentials="my-aws-creds-block"
    )
s3bucket.save(name="my-s3-bucket-block", overwrite=True)

Azure

Note that the AzureBlobStorageCredentials block is not the same as the Azure block that ships with Prefect. The AzureBlobStorageCredentials block used in this example is part of the prefect-azure library and provides additional functionality.

Azure blob storage doesn’t require a separate block. The connection string used in the AzureBlobStorageCredentials block can encode the information needed.

GCP

Note that the GcsBucket block is not the same as the GCS block that ships with Prefect. The GcsBucket block is part of the prefect-gcp library and provides additional functionality.

Next, reference the credentials block created above.

from prefect_gcp.cloud_storage import GcsBucket

gcsbucket = GcsBucket(
    bucket="my-bucket-name", 
    credentials="my-gcp-creds-block"
    )
gcsbucket.save(name="my-gcs-bucket-block", overwrite=True)

Run the code to create the block. You should see a message that the block was created.

Write data

Use your new block inside a flow to write data to your cloud provider.

AWS

from pathlib import Path
from prefect import flow
from prefect_aws.s3 import S3Bucket

@flow()
def upload_to_s3():
    """Flow function to upload data"""
    path = Path("my_path_to/my_file.parquet")
    aws_block = S3Bucket.load("my-s3-bucket-block")
    aws_block.upload_from_path(from_path=path, to_path=path)

if __name__ == "__main__":
    upload_to_s3()

Azure

from prefect import flow
from prefect_azure import AzureBlobStorageCredentials
from prefect_azure.blob_storage import blob_storage_upload

@flow
def upload_to_azure():
    """Flow function to upload data"""
    blob_storage_credentials = AzureBlobStorageCredentials.load(
        name="my-azure-creds-block"
    )

    with open("my_path_to/my_file.parquet", "rb") as f:
        blob_storage_upload(
            data=f.read(),
            container="my_container",
            blob="my_path_to/my_file.parquet",
            blob_storage_credentials=blob_storage_credentials,
        )

if __name__ == "__main__":
    upload_to_azure()

GCP

from pathlib import Path
from prefect import flow
from prefect_gcp.cloud_storage import GcsBucket

@flow()
def upload_to_gcs():
    """Flow function to upload data"""
    path = Path("my_path_to/my_file.parquet")
    gcs_block = GcsBucket.load("my-gcs-bucket-block")
    gcs_block.upload_from_path(from_path=path, to_path=path)

if __name__ == "__main__":
    upload_to_gcs()

Read data

Use your block to read data from your cloud provider inside a flow.

AWS

from prefect import flow
from prefect_aws import S3Bucket
from prefect_aws.s3 import S3Bucket

@flow
def download_from_s3():
    """Flow function to download data"""
    s3_block = S3Bucket.load("my-s3-bucket-block")
    s3_block.get_directory(
        from_path="my_path_to/my_file.parquet", 
        local_path="my_path_to/my_file.parquet"
    )

if __name__ == "__main__":
    download_from_s3()

Azure

from prefect import flow
from prefect_azure import AzureBlobStorageCredentials
from prefect_azure.blob_storage import blob_storage_download

@flow
def download_from_azure():
    """Flow function to download data"""
    blob_storage_credentials = AzureBlobStorageCredentials.load(
        name="my-azure-creds-block"
    )
    blob_storage_download(
        blob="my_path_to/my_file.parquet",
        container="my_container",
        blob_storage_credentials=blob_storage_credentials,
    )

if __name__ == "__main__":
    download_from_azure()

GCP

from prefect import flow
from prefect_gcp.cloud_storage import GcsBucket

@flow
def download_from_gcs():
    gcs_block = GcsBucket.load("my-gcs-bucket-block")
    gcs_block.get_directory(
        from_path="my_path_to/my_file.parquet", 
        local_path="my_path_to/my_file.parquet"
    )

if __name__ == "__main__":
    download_from_gcs()

Next steps

Check out the prefect-aws, prefect-azure, and prefect-gcp docs to see additional methods for interacting with cloud storage providers. Each library also contains blocks for interacting with other cloud-provider services.