is that intended? if so, is there a way to skip the SKIP signals?

I feel like this must be pretty basic: Is there a way to combine/string together multiple flows?

How do I map over the product of two parameters? I consulted the docs and issues (notably https://github.com/PrefectHQ/prefect/issues/1311 and https://github.com/PrefectHQ/prefect/issues/674), but still couldn't figure it out. Some attempts: 1) Passing a task into a "compose" task as a parameter --> result = compose.map(ids, unmapped(generate_task), unmapped(dates)) --> got

TypeError: generate_task() missing 2 required positional arguments:

2) Calling a known task from a "compose" task --> got

ValueError: Could not infer an active Flow context.

3) Typecasting parameter to list --> Parameter is not iterable

Related to my above question, how do I access the value of a parameter? For example, I want to use

flattened_lists = itertools.chain(*nested_lists_param)

but can't iterate through the Parameter itself.

Dask question here: I am new to distributed systems, and want to run a large number of tasks (N = 10k+) in parallel. Afaik, Prefect+Dask is scalable to 10k+ parallel tasks (although I know it's ideal to batch many small tasks to few longer-running tasks if possible). However, trying different small toy tasks (and various combinations of

--nprocs

--nthreads

on workers), it seems that consistently - Running Client() locally starts immediately, as expected. - Running with one worker server, starting takes much longer as N increases For large N:

[2019-09-30 22:59:15,781] INFO - prefect.TaskRunner | Task 'stage_0': Starting task run...
distributed.utils_perf - INFO - full garbage collection released 561.93 MB from 0 reference cycles (threshold: 10.00 MB)
distributed.core - INFO - Event loop was unresponsive in Worker for 34.31s.  This is often caused by long-running GIL-holding functions or moving large chunks of data. This can cause timeouts and instability.

- Running on multiple servers, the Dask scheduler also takes much way longer than the expected ~1ms overhead/task, before any activity happens (on the client servers or the Dask scheduler dashboard). Why is scheduling taking so long? How could I adjust scheduling policy (https://distributed.dask.org/en/latest/scheduling-policies.html) to speed up running large number of small tasks?

Hi folks, I'm trying Prefect the first time and having a few questions regarding best practices that I didn't get a hold of in the documentation. 1. Logging: I'm extracting ~10k IDs out of a database and return that as a list of a task. The next task will map over it and continue working on them. However that list is printed in the logs completely, but I don't see value for that big chunk of log. Are there ways of disabling printing the result of a task in the logs for a single task without disabling logging in general? Or am I misusing tasks and should better return a generator instead of a list? 2. How to integrate a Prefect Flow into a CLI? I currently have a Python CLI tool that takes a database connection string (where to get the IDs from mentioned above) and processes this data with a plain loop and want to migrate that to Flow. How do I now call the flow in the CLI script in order to make the CLI script behave as a CLI should, most importantly the a exit code should be not 0 when the flow state is not successfull. Is there a guide for best practices of using prefect in CLIs?

Hi guys! The project looks great so far and me and my team are investigating the possibility of using it instead of airflow. I looked through the docs and skimmed through the code and code not find any authentication/access management like the kind that airflow has in prefect-core. Is this correct? and is there any plan to implement in the open sourced project? Thanks a lot for your time and I am excited to see how the project evolves!

If a Parameter consists of e.g. a list of length 100k, at what stage and how may it be batched with dask.Bag?

Hey all, I have been using local secrets in my production flows for a while. I want to use some other key-value store for my secrets, as it is the secure and intended way. I haven’t found much in the docs, how would I be telling Prefect Core where and how it should access this new

Secrets

store, say AWS Secrets Manager

Hey all, is there any way for a Task to access the result of another Task (in the same flow) aside from passing the upstream task to the downstream task's kwargs?

Is there a way to track data dependencies across nodes per a data lineage graph?

hey guys, absolutely love the work that you are doing... huge fan of Prefect, i see it being the future of workflow engines... quick question - are you guys at all considering the implementation of a CLI?

Hi again everyone, our company has our own logger class (based on the excellent Loguru) which we use as a drop in replacement for the standard logger. Is there a way to use it in Prefect? Can I replace Prefect’s logger with ours? Or redirect Prefect’s logger to ours as a handler? Thank you again all.

How does Prefect compare to Kubeflow?

also, does Prefect have an extension point to build custom Tasks?

Just saw the announcement of the Prefect scheduler! That is awesome! Congrats on getting that out! Just so I’m understanding it, is the main use of it as a “trial” type of service of the Cloud offering? Or would it be used to help roll your own infra and work with Prefect? (Tried to read through as much as I could, site isn’t very mobile friendly)

Hi - I ran into an error with the visualizer python 3.7.4 The

flow

did run correctly? any suggestions?

hello prefect, long time listener first time caller, is somebody working on Azure tasks that wants to answer a few questions

There is a way to append one Flow to another? My problem is the following: I have a flow for computing data and a flow for computing data and plotting it. I'm wondering if there is a way to do this without creating two flows with repeated code. Thank you!

last question, and apologies if this has been answered. if folks have an existing Dask-ified task but want to use other parts of Prefect with the DaskExecutor, is it recommended to rewrite the existing task using just prefect or simply run the old task separately

Hello team! I'm facing a weird error that I'm not sure if related to Prefect or not: I'm invoking a lambda inside a task and getting the following error:

[2019-10-08 17:26:55,377] ERROR - prefect.TaskRunner | Unexpected error: TooManyRequestsException('An error occurred (TooManyRequestsException) when calling the Invoke operation (reached max retries: 4): Rate Exceeded.')
Traceback (most recent call last):
  File "/usr/local/lib/python3.7/site-packages/prefect/engine/runner.py", line 48, in inner
    new_state = method(self, state, *args, **kwargs)
  File "/usr/local/lib/python3.7/site-packages/prefect/engine/task_runner.py", line 869, in get_task_run_state
    self.task.run, timeout=self.task.timeout, **raw_inputs
  File "/usr/local/lib/python3.7/site-packages/prefect/utilities/executors.py", line 79, in timeout_handler
    return fn(*args, **kwargs)
  File "/Users/aneto/Documents/github/data-platform/scheduler/main.py", line 244, in run_sync_loads
    Payload=json.dumps(load)
  File "/usr/local/lib/python3.7/site-packages/botocore/client.py", line 357, in _api_call
    return self._make_api_call(operation_name, kwargs)
  File "/usr/local/lib/python3.7/site-packages/botocore/client.py", line 661, in _make_api_call
    raise error_class(parsed_response, operation_name)
botocore.errorfactory.TooManyRequestsException: An error occurred (TooManyRequestsException) when calling the Invoke operation (reached max retries: 4): Rate Exceeded.

The function has a 15min timeout and it is being called as:

client = boto3.client('lambda')
sync_results = client.invoke(
            FunctionName='myfunction',
            Payload=json.dumps(load)
        )

The code following this invocation is not being called. I noticed that even before the Prefect retry the function is being called more than once.

Hi Prefects, I’ve run into a problem when trying to

merge

two conditional branches back together, if the result of the previous step was a Pandas dataframe you get these errors;

[2019-10-09 09:27:02,948] INFO - prefect.TaskRunner | Task 'Merge': Starting task run...
[2019-10-09 09:27:02,961] ERROR - prefect.TaskRunner | Unexpected error: ValueError('The truth value of a DataFrame is ambiguous. Use a.empty, a.bool(), a.item(), a.any() or a.all().')
...
[2019-10-09 09:27:02,965] INFO - prefect.TaskRunner | Task 'Merge': finished task run for task with final state: 'Failed'

Presumably this is because of the

Merge

Task comparing the result to the

NoResult

object.

Is there something I can do instead of

merge

, perhaps my own task to determine which of a variadic number of results I want to pass on?

Hi - Is there a way to

pass

on a if else branch?

Perhaps I’m just a bit dense but as per the docs JinjaTemplate() should take **kwargs (https://docs.prefect.io/api/unreleased/tasks/strings.html#jinjatemplate). Yet when I pass it keyword arguments it complains about super().__init__() getting unexpected keyword arguments. Bug?

Hi! Is there a way to use checkpointing as a persistent cache? It doesn't seem to load the checkpointed file by default

Hi prefect team, with the slack notifier is it possible to get the parameters that the task was called with passed through also?

Heya. Project looks awesome. I've been messing with RAPIDS, which is also kinda part of the Daskverse (basically a way to painlessly use GPUs from the Python data stack). Would Prefect be able to easily interface with it? https://rapids.ai/dask.html