Test Dask setup with Prefect and AWS EKS on Fargate

dask-k8.py

from prefect.environments.storage import Docker
from prefect.environments import DaskKubernetesEnvironment
from prefect import task, Flow
import random
from time import sleep


@task
def inc(x):
    sleep(random.random() / 10)
    return x + 1


@task
def dec(x):
    sleep(random.random() / 10)
    return x - 1


@task
def add(x, y):
    sleep(random.random() / 10)
    return x + y


@task(log_stdout=True)
def list_sum(arr):
    return sum(arr)


with Flow("dask-k8") as flow:
    random.seed(123)
    incs = inc.map(x=range(100))
    decs = dec.map(x=range(100))
    adds = add.map(x=incs, y=decs)
    total = list_sum(adds)

if __name__ == '__main__':
    flow.storage = Docker(registry_url="<YOUR_ECR_REGISTRY_ID>.dkr.ecr.eu-central-1.amazonaws.com", image_tag='latest')
    flow.environment = DaskKubernetesEnvironment(min_workers=3, max_workers=5)
    flow.register(project_name="Medium_AWS_Prefect")

Hi, is this creating dask cluster dynamically? New to aws and dask.

Hi, yes. However, there has been a change in how environments are configured. Have a look at the docs:

• deprecated version: https://docs.prefect.io/orchestration/execution/dask_k8s_environment.html
• new way: https://docs.prefect.io/orchestration/flow_config/run_configs.html#run-configuration

Hope that helps!