Prefect Team

[2021-05-12 12:26:05-0700] INFO - prefect.task_name | Waiting for next scheduled run at 2021-05-13T10:00:00+00:00
2021-05-12 12:26:05,221 - prefect.task_name - INFO - Waiting for next scheduled run at 2021-05-13T10:00:00+00:00

Let's say that you want to always run a flow with parameters that are generated by another flow. Naively, you might manually wait for one flow to finish and then manually trigger the next flow to run with the appropriate parameter values. Prefect makes this pattern easy to automate via the StartFlowRun:

parameters

DaskKubernetesEnvironment

run_configs

executors

DaskExecutor

flow.storage = Docker(registry_url=ecr,
                      python_dependencies=python_dependencies,
                      files={f'{getcwd()}/src/dask_flow': '/modules/dask_flow'},
                              env_vars={"PYTHONPATH": "$PYTHONPATH:modules/"},
                              image_tag='latest')

flow.run_config=KubernetesRun()
flow.executor=DaskExecutor(
   cluster_class="dask_cloudprovider.aws.FargateCluster",
   cluster_kwargs={'n_workers': 5, 'region_name':aws_region})

client

client.upload_file)

ModuleNotFoundError: No module named 'aircraftlib'
[2021-05-13 09:27:29+0200] ERROR - prefect.etl | Unexpected error occured in FlowRunner: ModuleNotFoundError("No module named 'aircraftlib'")

helm install prefect prefecthq/prefect-server -f values.yaml

Error: YAML parse error on prefect-server/templates/hasura/deployment.yaml: error converting YAML to JSON: yaml: line 31: mapping values are not allowed in this context

securityContext:
  runAsUser: 1000
  runAsGroup: 1000

securityContext: {}

securityContext:
    {{- toYaml . | nindent 8 }}
  {{- end }}

0.14.17

0.14.19

Traceback (most recent call last):	
File "/usr/local/bin/dask-scheduler", line 5, in <module>	
from distributed.cli.dask_scheduler import go	
File "/usr/local/lib/python3.8/site-packages/distributed/cli/dask_scheduler.py", line 120, in <module>	
def main(	
File "/usr/local/lib/python3.8/site-packages/click/decorators.py", line 247, in decorator	
_param_memo(f, OptionClass(param_decls, **option_attrs))	
File "/usr/local/lib/python3.8/site-packages/click/core.py", line 2467, in __init__	
super().__init__(param_decls, type=type, multiple=multiple, **attrs)	
File "/usr/local/lib/python3.8/site-packages/click/core.py", line 2106, in __init__	
raise ValueError(	
ValueError: 'default' must be a list when 'multiple' is true.

Failed to load and execute Flow's environment: ModuleNotFoundError("No module named '/root/'")

IntervalSchedule

now().minute == 30

now().minute % 5 == 0

from prefect.agent.ecs import ECSAgent

agent = ECSAgent(
    name='Test Agent',
    region_name='eu-west-1',
    cluster='my-test-cluster',
    launch_type='FARGATE',
    run_task_kwargs_path='./config.yaml'
)
agent.start()

[2021-05-14 12:46:54,069] ERROR - Test Agent | Failed to infer default networkConfiguration, please explicitly configure using `--run-task-kwargs`
Traceback (most recent call last):
  File "/Users/kha/Desktop/foo/agent.py", line 3, in <module>
    agent = ECSAgent(
  File "/opt/homebrew/Caskroom/miniforge/base/envs/zoined/lib/python3.9/site-packages/prefect/agent/ecs/agent.py", line 234, in __init__
    ] = self.infer_network_configuration()
  File "/opt/homebrew/Caskroom/miniforge/base/envs/zoined/lib/python3.9/site-packages/prefect/agent/ecs/agent.py", line 274, in infer_network_configuration
    raise ValueError(msg)
ValueError: Failed to infer default networkConfiguration, please explicitly configure using `--run-task-kwargs`

import prefect
from prefect import task, Flow
from prefect.run_configs import ECSRun


@task
def hello_task():
    logger = prefect.context.get("logger")
    <http://logger.info|logger.info>("Hello World!")
    return 0


task_definition = {
    'networkMode': 'awsvpc',
    'cpu': 1024,
    'memory': 2048,
    'containerDefinitions': [
        {'name': 'flow'}
    ]
}


run_config = ECSRun(
    image='prefecthq/prefect',
    task_definition=task_definition
)

flow = Flow("hello-flow",
            tasks=[hello_task],
            run_config=run_config)

flow.register(project_name='Prefect_Tutorials', labels=['aws', 'ecs', 'fargate'])

Failed to load and execute Flow's environment: ModuleNotFoundError("No module named '/Users/kha/'")

prefect register flow_name --label=staging --no-schedule

from prefect import task, Flow
from prefect.tasks.secrets import PrefectSecret
from prefect.tasks.databricks import DatabricksRunNow
import requests


conncfg = {"host":"<https://xxx.cloud.databricks.com/>", "token":"xxx"}


job_json = {
      "job_id": 24242,
      "notebook_params": {}
    }


@task
def run_databricks(job_config):
    notebook_run = DatabricksRunNow(json=job_config).run()
    run_id = notebook_run(databricks_conn_secret=conncfg)
    headers = { "Authorization" : "Bearer xxx" }
    r = requests.get('<https://xxx.cloud.databricks.com/api/2.0/jobs/runs/get-output?run_id=%s'%(runid)>, headers=headers)
    results = r.json()
    output =  results['metadata']['notebook_output']
    if output.has_key('message'):
        raise Exception(output['message'])



with Flow('testflow') as flow:
    run_databricks(job_json)

Unexpected error: AssertionError('A databricks connection string must be supplied as a dictionary or through Prefect Secrets')