example_s3_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

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