HdfsResult

prefect

location

Result

location

exists(...)

read(...)

location

exists

read

prefect

Result

Result.read(…)

self.value

from prefect import Flow, task, apply_map

@task
def task1(arg1):
    return arg1

@task
def task2(arg2):
    return sum(arg2)

def jobMap(arg):
    print("branch"+str(arg))
    arg2 = task1.map(range(2))
    task2(arg2)

with Flow("workflow") as flow:
    apply_map(jobMap, range(3))
        
flow.run()

mapped=True

LocalDaskExecutor

LocalExecutor

d6962d8d-f4b0-4e04-8217-82d41400a043

repo_name/projects/npa_project/npa_project_module/npa_flow.py

npa_project_module

prefect run -m npa_project_module.npa_flow

No module named "npa_project_module"

npa_project

npa_project_module

class SetupStorageTask(Task): 
    def run(self, mode) -> Storage:    
        return Storage(mode=mode)

setup_storage_task = SetupStorageTask()
storage = setup_storage_task(mode='rw')

class SetupStorageTask(Task):

    def __init__(self, mode):
        self.mode = mode
    
    def run(self) -> Storage:
        return Storage(mode=self.mode)

setup_storage_task = SetupStorageTask(mode='rw')
storage = setup_storage_task()

or can be just
storage = SetupStorageTask(mode='rw')

Dockerfile

create_flow_run

run docker

query {
  flow(
    where: {name: {_eq: "YYY"}, project: {name: {_eq: "XXX"}}, flow_runs: {state: { _eq: "Success"}}}
  ) {
    flow_runs_aggregate{
      aggregate {
        count
      }
    }
  }
}

HdfsSparkDataFrameResult

prefect

@task(result=HdfsSparkDataFrameResult(…))

Result

TestOutputCaching

EvaluateModelOnMetric(model: Model = …, metric: Metric = …)

models: List[Model] = [model_a, model_b, model_c]

metrics: List[Metric] = [precision, recall, f1_score]

EvaluateModelsOnMetrics(models: List[Model] = …, metrics: List[Metric] = …)

itertools.product

EvaluateModelOnMetric

ECSAgent

poviderStrategy

No Container Instances were found in your cluster.

LaunchType

An error occurred (InvalidParameterException) when calling the RunTask operation: You may choose a capacity provider or a launch type but not both.

prefect.task.airbyte

#14 1.604 Traceback (most recent call last):
#14 1.604   File "flow/etl.py", line 7, in <module>
#14 1.604     from prefect.tasks.airbyte import AirbyteConnectionTask
#14 1.604 ModuleNotFoundError: No module named 'prefect.tasks.airbyte'

schedule = Schedule(
        clocks=[clocks.IntervalClock(timedelta(days=1))],
        # filters=[filters.at_time(time(10))],
        or_filters=[filters.between_times(pendulum.time(9), pendulum.time(9))]
    )

FilterTask

State payload is too large

result=None

Failed to set task state with error: ClientError([{'path': ['set_task_run_states'], 'message': 'State payload is too large.', 'extensions': {'code': 'INTERNAL_SERVER_ERROR'}}])
Traceback (most recent call last):
  File "/usr/local/lib/python3.8/site-packages/prefect/engine/cloud/task_runner.py", line 91, in call_runner_target_handlers
    state = self.client.set_task_run_state(
  File "/usr/local/lib/python3.8/site-packages/prefect/client/client.py", line 1917, in set_task_run_state
    result = self.graphql(
  File "/usr/local/lib/python3.8/site-packages/prefect/client/client.py", line 569, in graphql
    raise ClientError(result["errors"])
prefect.exceptions.ClientError: [{'path': ['set_task_run_states'], 'message': 'State payload is too large.', 'extensions': {'code': 'INTERNAL_SERVER_ERROR'}}]

trigger=any_successful

from prefect import Flow
from prefect.schedules import CronSchedule
from prefect.tasks.prefect import create_flow_run, wait_for_flow_run


weekday_schedule = CronSchedule(
    "30 9 * * 1-5", start_date=pendulum.now(tz="US/Eastern")
)


with Flow("parent-flow", schedule=weekday_schedule) as flow:
    
    # assumes you have registered the following flows in a project named "examples"
    flow_a = create_flow_run(flow_name="A", project_name="examples")
    wait_for_flow_a = wait_for_flow_run(flow_a, raise_final_state=True)
    
    flow_b = create_flow_run(flow_name="B", project_name="examples")
    wait_for_flow_b = wait_for_flow_run(flow_b, raise_final_state=True)
    
    flow_c = create_flow_run(flow_name="C", project_name="examples")
    wait_for_flow_c = wait_for_flow_run(flow_c, raise_final_state=True)
    
    flow_d = create_flow_run(flow_name="D", project_name="examples")
    wait_for_flow_d = wait_for_flow_run(flow_d, raise_final_state=True)
    
    b = wait_for_flow_b(upstream_tasks=[wait_for_flow_a])
    c = wait_for_flow_c(upstream_tasks=[wait_for_flow_a])
    d = wait_for_flow_d(upstream_tasks=[b, c])
flow.run()

@task
def check_cond():
    return False


@task
def run_task(anchor_date, days_from):

    logger = prefect.context.logger
    logger.critical(f"{days_from=}")


with Flow('foo') as flow:

    anchor_date = Parameter(name="anchor_date", default=None)
    cond = check_cond()

    with case(cond, True):
        anchor_date = run_task(anchor_date, days_from=0)

    with case(cond, False):
        anchor_date = run_task(anchor_date, days_from=-1)

False

check_cond

True