example_s3_multilevel_mapping_flow.py

from prefect import Flow, task, Parameter


@task
def get_s3_keys(bucket: str):
    # get a list_of_keys from your s3 bucket with boto3
    return list_of_keys

@task
def process_s3_key(key: str):
    # do something to each of the files

@task
def process_s3_key_even_more(key: str):
    # do something else

@task
def process_s3_key_again(key: str):
    # do yet another thing to this s3 file

with Flow("process_bucket") as flow:
    bucket = Parameter('bucket')
    s3_keys = get_s3_keys(bucket)
    process_s3_key_again.map(process_s3_key_even_more.map(process_s3_key.map(s3_keys)))