Quick question about the prefect CLI. It seems that when I run a command like

prefect describe flows --name foo-bar

That I get output like this, which includes information about the storage and parameters, but contains no information about the run configuration (I’m using kubernetes)

{
  "archived": false,
  "created": "2021-04-20T19:01:08.156447+00:00",
  "description": null,
  "environment": null,
  "name": "inference-alpha-poc",
  "parameters": [
    .......
  ],
  "project": {
    "name": "recursion-mlops-poc"
  },
  "storage": {
    "__version__": "0.14.16",
    "access_token_secret": "GITHUB_ACCESS_TOKEN",
    "base_url": null,
    "flows": {
      "inference-alpha-poc": "prefect_cloud/inference_flow.py"
    },
    "path": "prefect_cloud/inference_flow.py",
    "ref": "prefect-cloud",
    "repo": "recursionpharma/inference-alpha",
    "secrets": [],
    "type": "GitHub"
  },
  "version": 9
}

Is this expected, or a bug? Why isn’t the run configuration shown in this output?

I have: • task1 mapped over list1 and task2 mapped over list2 • task2 isn't mapped over the output of task1 • I don't want task2 to start until after task1 has completed. I tried adding

upstream_tasks=[task1]

to task2. My gut told me this probably wasn't going to do what I wanted, but it was worth a try. My gut was right as task2 only process the number of elements in list2 that are in list1.

import prefect
from prefect import Flow, task


@task
def mapped(value: int) -> int:
    prefect.context.get("logger").info(f"{value}")
    return value


if __name__ == "__main__":
    with Flow("mapped_upstream") as flow:
        list1 = [1, 2]
        task1 = mapped.map(value=list1)

        list2 = [10, 20, 30]
        task2 = mapped.map(value=list2, upstream_tasks=[task1])
    flow.run()

What is the correct way to get mapped task2 to have an upstream dependency on task1? [2021-04-28 143831-0600] INFO - prefect.FlowRunner | Beginning Flow run for 'mapped_upstream' [2021-04-28 143831-0600] INFO - prefect.TaskRunner | Task 'mapped': Starting task run... [2021-04-28 143831-0600] INFO - prefect.TaskRunner | Task 'mapped': Finished task run for task with final state: 'Mapped' [2021-04-28 143831-0600] INFO - prefect.TaskRunner | Task 'mapped[0]': Starting task run... [2021-04-28 143831-0600] INFO - prefect.mapped[0] | 1 [2021-04-28 143831-0600] INFO - prefect.TaskRunner | Task 'mapped[0]': Finished task run for task with final state: 'Success' [2021-04-28 143831-0600] INFO - prefect.TaskRunner | Task 'mapped[1]': Starting task run... [2021-04-28 143831-0600] INFO - prefect.mapped[1] | 2 [2021-04-28 143831-0600] INFO - prefect.TaskRunner | Task 'mapped[1]': Finished task run for task with final state: 'Success' [2021-04-28 143831-0600] INFO - prefect.TaskRunner | Task 'mapped': Starting task run... [2021-04-28 143832-0600] INFO - prefect.TaskRunner | Task 'mapped': Finished task run for task with final state: 'Mapped' [2021-04-28 143832-0600] INFO - prefect.TaskRunner | Task 'mapped[0]': Starting task run... [2021-04-28 143832-0600] INFO - prefect.mapped[0] | 10 [2021-04-28 143832-0600] INFO - prefect.TaskRunner | Task 'mapped[0]': Finished task run for task with final state: 'Success' [2021-04-28 143832-0600] INFO - prefect.TaskRunner | Task 'mapped[1]': Starting task run... [2021-04-28 143832-0600] INFO - prefect.mapped[1] | 20 [2021-04-28 143832-0600] INFO - prefect.TaskRunner | Task 'mapped[1]': Finished task run for task with final state: 'Success' [2021-04-28 143832-0600] INFO - prefect.FlowRunner | Flow run SUCCESS: all reference tasks succeeded

Is there a nature way to work with Parameters? e.g the example below will throw an exception because for_date has type Parameter

with Flow("Demo") as flow:
    for_date = Parameter('for_date', required=True, default='20200101')
    for_date_str = dt.datetime.strptime(for_date, "%Y%m%d")

Hi guys, is there a Made with Prefect GitHub badge for illustrating repos ?

Hi @Kyle Moon-Wright, just following up from the email. You mentioned the interval kwarg. I know where this is in the code, but, was there a specific part of the UI you were seeing this odd value? Or, where did you see it?

Hey guys is this a prefect compliant way of defining a task?

task(
   fn=lambda a, b: a+b,
   name='some_task'
)(some_a, some_b)

Hello Team, I have a query regarding upstream tasks,

Hello team, Is there a bug in "apply_map" working with flatten? I tried to modify the example from the doc (https://docs.prefect.io/core/concepts/mapping.html#flat-mapping) to use apply_map:

from prefect import Flow, task, flatten, apply_map
@task
def A():
    return [1, 2, 3]

@task
def B(x):
    return list(range(x))

@task
def C(y):
    return y + 100
    
def foo(y):
    return C(y)

with Flow('flat map') as f:
    a = A() # [1, 2, 3]
    b = B.map(x=a) # [[0], [0, 1], [0, 1, 2]]
    c = apply_map(foo, y=flatten(b)) # [100, 100, 101, 100, 101, 102]

While running it, I get : "TypeError: can only concatenate list (not "int") to list" Am I missing something?

Hey Team, I have a Kubernetes config question: I setup a k8s prefect agent running in a few clusters. It is pretty straight forward and the prefect commandline tool will spit out the yaml to kubectl to create a k8s Deployment running the agent. The agent then polls the cloud instance for new jobs and spins up new k8s Jobs/Pods that are running the tasks. Given I have a number of different AWS VPCs which all have AWS VPC Endpoints pointing to the same services but with different IPs mapped to them I would like to assign different HostAlias mappings so that same name in multiple clusters can be assigned to different IP addresses. I know it is possible since I have a cluster configured in this way but the guy setting it up has since left. Example output from kubectl in attached file with ******* marking the place of interest and descriptions. Any pointers are of great interest!!!

Hello, I'm new to Prefect and I'm wondering if I can use Prefect for my use case. I have a python script that extracts data from Google sheets and loads them into SnowFlake. A cron is used to schedule it running daily. Similarly, there are singer taps and targets orchestrated by the pipelinewise to load data from databases to SnowFlake. They are also scheduled by crons. The loaded data are then transformed using DBT and are scheduled much later than the above crons. Most of the examples I see in the Internet are workflows running python functions with task decorator.

Hi,. I am still pretty confused. I am new to the prefect. I need to have run an ETL task. Such that Extract is done by one agent and transform /load is done by another agent . How can I achieve that. How can I run two agents. I know I can do labels with agents. But can you refer to any example. It would help to understand better, Please guide.

Hey all, While trying to run a flow via ECS agent, I get the following error:

Failed to load and execute Flow's environment: StorageError("An error occurred while unpickling the flow:\n TypeError('code() takes at most 15 arguments (16 given)')\n This may be due to one of the following version mismatches between the flow build and execution environments: \n - python: (flow build with '3.8.5', currently running with '3.7.10')")

I'm running prefect server, prefect agent and registration environment on 3 different Ubuntu machines, all running:

$ python3
Python 3.8.5 (default, Jan 27 2021, 15:41:15) 
[GCC 9.3.0] on linux
Type "help", "copyright", "credits" or "license" for more information.
>>> import sys
>>> import prefect
>>> prefect.__version__
'0.14.17'
>>> sys.version_info
sys.version_info(major=3, minor=8, micro=5, releaselevel='final', serial=0)

I don't understand where does the agent pick up the python 3.7.10 version, and why is the task executed with that environment (the python 3.7 is not even installed on the server nor agent environment). Is there a way to set the python version on the execution environment, or am I missing something?

Did the url for the documentation on Cloud Hooks change recently? I'm getting a 404 when I go to this page that I had bookmarked not too long ago

Quite an aws specific question.. I have a batch job (created before I started using Prefect) which is calling a lambda function approx 1000 times, each time with slightly different args (it is processing images on S3). The lambdas happen to all request the same file on S3 (used as a reference image) and I was getting S3 rate limits, so taking the lazy route I just throttled the lambdas by running them in serial using

RequestResponse

. I also log the response from the lambdas to a db, currently at the end of the batch job. My question is whether I am better to stick with running batch job from Prefect, or if I am better to use Prefect to call the 1000 lambdas in parallel using lambda

Event

mode, then use Prefects ability to handle retries to manage the show..? Also a related question is when is the best time to log to the db. Thanks in advance! ps loving prefect way more than airflow

Hi All, We have a handy little Task class called 

api_data_to_s3

 that sends json data to S3. In other flows, we import this class and pass it two arguments, the vendor name and the api_endpoint (ex. 

api_data_to_s3(vendor="google", api_endpoint="keywords")

 . The issue is that some vendors have multiple endpoints and when we try to parameterize the api_endpoint we get some errors. I'll post some code in the thread below but I'd love to know the recommended way to solve.

I'm currently working on using the prefect API through an iPython environment (Jupyter Notebook atm). Because my environment is heavily containerized, I'm looking at using the Webhook Storage, since that'll let me reuse a bunch of existing infrastructure. Am I correct in my reading of the source that if I want to store the flow as a script (i.e. not have it pickled), I would need to write a file out temporarily to local storage, if the flow is some python functions defined in a cell?

Will Prefect allow me to access the built in functions of the data type being returned by the Task or does the Task object occlude some of those calls? Specifically for dictionaries, can I do somethin glike this:

d_task  = task(return some_dict)
m_task = task(lambda x: do_map_stuff(x))

with Flow f as ('flow'):
  some_data = d_task()
  mapped_result = m_task.map(some_data.values())

Hey guys, I build modules with tasks that I'm using in my flow, but updating flow version these modules aren't updated. What are the best practices?

Hi all, for an instance of Prefect Server, what are the recommended specs for a system to run Prefect effectively?

I am getting this warning and I don't understand what it means - UserWarning: Task <Task: pack_ds_data> has retry settings but some upstream dependencies do not have result types.

Hi, Can I run two Local Agents using the same machine. I am trying to test my code and I want to assign my flows to different agents. So I am trying to run two agents on my machine

Hi everyone, I just realized something about

merge

, and I find it kind of unexpected. Here goes: Usually, when using something like

ifelse

and

merge

, the flow goes like:

skip_if_true -> actual_task_for_false -> merge
skip_if_false -> actual_task_for_true -> merge

This works as expected. However I tried to directly merge tasks that can or cannot

SKIP

:

skippable_1 = do_i_skip(skip=True)
skippable_2 = do_i_skip(skip=False)
merge(skippable_1, skippable_2)

merge

results in the

SKIP

signal raised in skippable_1, rather than skippable_2. That is unexpected, at least for me. Is this expected behavior? Can I somehow connect these explicit

SKIP

ping tasks with a merge directly?

Hello, having an issue with passing parameters/context from flow to flow:

@task
def log_scheduled_start_time():
    """Print scheduled_start_time in logger."""
    logger = prefect.context.get("logger")
    scheduled_start_time = prefect.context.get("scheduled_start_time")
    flow_name = prefect.context.get("flow_name")
    <http://logger.info|logger.info>(f"{flow_name}: {scheduled_start_time}")


with Flow("subflow") as subflow:
    log_scheduled_start_time()

subflow.register(project_name="test_schedule", labels=["test"])

with Flow("mainflow") as mainflow:
    log_scheduled_start_time()
    StartFlowRun(
        flow_name="subflow", project_name="test_schedule", wait=True,
        scheduled_start_time=prefect.context.get("scheduled_start_time")
    )

mainflow.register(project_name="test_schedule", labels=["test"])

Expectation: mainflow's

scheduled_start_time

is passed down to subflow, both flows log the same time. Reality: subflow starts without a

scheduled_start_time

, defaults to

now()

, logged times are different. Use case: Subflow depends on

scheduled_start_time

to query some database, the agent that was supposed to run mainflow is down, mainflow and subflow are run a day later, subflow's query is then incorrect. Bonus question: I can't seem to find documentation on how to pass parameters from one flow to another/use a flow's results as parameters for another 🙇

Hey everyone. I'm running into some trouble running a prefect flow. It's a very simple one (converting json files to parquet). The problem is that when I try to run the flow (flow.run(...)) the flow doesn't work. I get to seet the "Saving status ..." log but the df.to_parquet never starts processing (looking at the dask dashboard). And it finally finishes with exception below. But when I run the task directly (to_parquet.run(...)) setting up the local dask cluster by myself it works perfectly. It's probably a rookie mistake but do you have any ideas what I'm doing wrong here? I'm using prefect 0.14.7 and dask 2014.4.1

Greetings, I'm having an issue with passing a Parameter from the command line to my flow, if I'm correct on the server the Parameter can be passed in the UI, but for testing I need to pass in sys.argv[2]. here's the relevant code:

def register__flow():

    

with Flow("FlowName", schedule=None, ) as flow:

        

file_name = Parameter(name='file_name',default=None)

        

params = get_params()

        

p = get_file_blob(

                        

file_name=f"notebooks/{file_name}",

                        

con_string=params["uploads_blob"],

                        

container="uploads"

        

)

        

flow.add_edge(file_name,params)

        

flow.add_edge(params, p)

    

if len(sys.argv) > 1 and sys.argv[1] == "register":

        

flow.register(project_name="project_x")

    

elif len(sys.argv) > 1 and sys.argv[1] == "test" and len(sys.argv[2]) > 1:

        

flow.run(parameters={'file_name':sys.argv[2]})

    

else:

        

print("Please use a quoted string of the file name...")

if __name__ == "__main__":

    

register__flow()

Thanks

Is there a canonical way of using a

PrefectSecret

to add a new item to the

prefect.context

? I see that you can add to the global context by doing something like:

with prefect.context(key='abc'):
    flow.run()

but it's not clear to me how to actually supply this value to the runtime context when my agents are deploying the flows and I'm not explicitly calling

flow.run()

. For context, I'm reading through the documentation on adding Slack handlers and I see that all you need to do is set a

SLACK_WEBHOOK_URL

secret in order for the buiilt-in

slack_notifier

to be able to post to your channel.

Hello all, I have been reading about github storage and had a few questions regarding it's use: currently we use docker images that we build with docker-compose and have a minimal task to create, start, and gather the logs. This has the disadvantage of not having logs in the Prefect UI and I was investigating github storage as a way to allow us to have logs in the UI. My question is, with using the github storage, what is the minimum that I need in the flow.py file for it to pull down the actual flow from github? And does it automatically create tasks based off of the tasks contained in the flow stored in github? I am trying to wrap my head around how it works and appreciate any advice and guidance with this

Maybe this is more than a prefect question. I use prefect to run batch job every day. Usually I have a lot of files like this: file_1, file_2, … file_100. Each processing task is independent to each other.  In order to parallelize the computation, I have several attempts: • Option 1: Single flow + Single agent + Dask + Dask workers: One flow processes multiple files. 1-on-1 mapping between task and file.

@task
process_file(x):

with Flow(xxx):
  [process_file(x) for x in file_names]

It works fine with LocalRun ◦ pro: Easy to set up, dask has the friendly api to control resource ◦ con: I have to set up the same running env for every worker added in this dask cluster.  • Option 2: Multiple flows + multiple agents: 1-on-1 mapping between flow and file e.g I could create 10 docker agents running on 10 hosts. Then in a script I create and run 100 flows, each flow processes one file. Let prefect distribute the flow for me ◦ pro: computation module is shipped with docker image, it is set up free. ◦ con: Not sure whether prefect is supposed to do the work load distribution duty. Even if yes, it is hard to control the resource consumption • Option 3: Single flow + K8s:   Build image for my computation module and register with K8S first. Within one flow, it create k8s tasks which request for creating 100 pods to process the file.  ◦ pro:  ▪︎ k8s could deal with the workload distribution. Adding nodes could be easy.  ▪︎ Any agent would be file as long as it could talk to k8s api ◦ con: complexity of setting up k8s? Appreciate for any thoughts and input here!

Hello, I'm trying to better understand the pricing/usage model. We have an ETL job that needs to make ~10k API calls to 2 different endpoints so ~20k calls total. We use Prefect's map and LocalDaskExecutor to utilize parallelization. However, if my understanding is correct, a single flow run counts as ~20k task runs and cost either

20,000 * 0.0025 = $50

or

20,000 * 0.005 = $100

depending on the plan you're on?

Hi is there a way to build a flow in a docker container without registering it? I can't seem to find it in the documentation but I'm trying to do something like this in my deploy script.

prefect build -p my_flow.py

which throws this usage error

Usage: prefect [OPTIONS] COMMAND [ARGS]...
Try 'prefect -h' for help.

Error: No such command 'build'.

Exited with code exit status 2