from prefect import task, Flow
from prefect.run_configs import ECSRun, DockerRun
from prefect.storage import S3
from prefect.executors import LocalDaskExecutor, DaskExecutor
import prefect
import time

@task(log_stdout=True)
def abc(x):
    time.sleep(5)
    <http://prefect.context.logger.info|prefect.context.logger.info>(x)
    print(x)
    return "hello"

with Flow("ecs_test", run_config=DockerRun(image="prefecthq/prefect:1.2.0-python3.7", env={"PREFECT__LOGGING__LOG_LEVEL": "DEBUG"}), 
                      executor = DaskExecutor(cluster_kwargs={"n_workers": 4, "threads_per_worker": 1})) as flow:
    abc(1)
    abc(2)

flow.storage = S3(bucket="coiled-prefect")
flow.register("databricks")