Hi folks, one pattern that we keep running into is using Parameters to override the result of a task. For example, we want to use some value from a database or the Parameter specified by the user.

with Flow("Update Data") as flow:
    last_export = get_last_export("my_table")

    update_from = Parameter("update_from", default=last_export)

    insert_query = parse_insert_query(update_from=update_from)

    run_execute_query(insert_query)

This doesn’t work but I hope it illustrates what I’m trying to do. Is there any way to do this without having to write an intermediate task to “merge” the two inputs? I realize this pattern is wasteful by running

get_last_export

for no reason if parameter is provided.

In what situation will I have to combine Prefect and Databricks? How Spark is related to them? And what do people create with Prefect, Databricks and Spark all together?

still confused on how we I would be able to transform map inputs in the following example:

@task
def taskA() -> List[int]: 
  return results
with Flow() as flow:
  param = Parameter('blah')
  a_results = taskA()
  flow_run_task = FlowRunTask()
  flow_run_task.map(flow_name=unmapped('DownstreamFlow'), parameters=[{'blah': param, 'resultId': result} for result in a_results])

in the example im doing the transformation with a list generator

but i don’t think i can actually loop over a_results inline inthe flow definition right?

Hi - I've noticed an issue recently in Prefect Cloud. If I try to change the state of a flow run, I get the following error. I just wanted to raise it as a fyi

Is there a feature yet to at the time of flow registration (via Python), be able to toggle off Heartbeats, Lazarus, etc.? If not, I would kindly request it!

👋 Hi, I am trying to unit-test my flows via a CI. How can I run a flow which has a schedule? (i.e. I need to run it once without triggering the schedule).

Hey folks, not sure if this is an issue with our settings but when setting a schedule in Cloud, the time is converted from UTC to our local timezone JST. So if I want the Flow to run at 11PM I have to set the schedule to run at the 14th hour. Is this expected behaviour?

Hey there, How can I change the port number of the prefect UI from the default

8080

? I have only seen a config example for the API port. Can you help me with a link to a page that contains all the possible config params for

config.toml

?

Hi folks, does anyone have any proof-of-concept examples of a project where there are multiple flows within the same project directory and some shared logic which the flows can import? I can get importing to work when running the flow locally with 

flow.run()

, but registering the flow on Prefect Server and triggering the flow from there fails because of module errors. I guess I'm just misunderstanding python's importing logic (I'm by no means an expert...) so a basic working example that I can build on would really help me get off the ground.

Hi Thanks a lot for the great product! I am working on porting my microscopy images analysis pipeline to prefect and currently I am not able to visualise custom logs in the UI (the CloudFlowRunner logs are present). I am running prefect server on a local cluster and connect via ssh to port 8080 and 4200 to connect to the UI and the apollo server from my laptop. I also spin an agent using the specific IP

prefect agent local start --api <http://172.22.0.5:4200>

I get IP from the terminal printout of the `prefect server start`command. Using this setup I am able to run flows, however I am not able to catch logs. I tested different server settings in the `config.toml`file but none fixed the issue:

[server]
 [server.ui]
 apollo_url="<http://0.0.0.0:4200/graphql>"

[server]
 [server.ui]
 apollo_url="<http://localhost:4200/graphql>"

If I understand how things work (please be patient I am a biologist 🙂 ) the issue is on my side and is caused by a mistake in the identification of the correct IP to use for setting up the system. If it is the case is there a way to predefine the IP where the services are starting? If I am wrong please let me know, any help is appreciated! Below is the toy flow I have been using to test test logging

import prefect
from prefect import task, Flow, Parameter, flatten, unmapped
from prefect.engine.executors import DaskExecutor
from prefect.utilities.debug import raise_on_exception
from prefect.utilities.logging import get_logger
from datetime import timedelta, datetime
from prefect.schedules import IntervalSchedule

# MOCK TASK FUNCTION TO BUY TIME
@task(task_run_name=lambda **kwargs: f"testing-logger-writing-logs-{kwargs['x']}-suiname",log_stdout=True)
def wlog(x):
    logger = prefect.context.get("logger")
    logger.debug('i am debugging')
    # logger = prefect_logging_setup('test')
    <http://logger.info|logger.info>(f'start sleep')
    time.sleep(20)
    <http://logger.info|logger.info>(f'done sleep')


a = list(range(10))
# with Flow("test_running",schedule=schedule) as flow:
with Flow("logging-flow",environment=LocalEnvironment(DaskExecutor(address='<tcp://193.10.16.58:18938>'))) as flow:
    logger = prefect.utilities.logging.get_logger()
    <http://logger.info|logger.info>('this log is generated in the flow')
    out_task = wlog.map(a)
    <http://logger.info|logger.info>('done')

flow.register(project_name="test")

Dear community, is there a simple

red button

to stop all running flows or tasks? 🔴 Our task concurrency limit indicates, that some tasks are still running, although no flow is running. So I guess there are some zombie tasks hiding somewhere, that I would like to see stopped. 😮 I am exploring the interactive API, but haven't found a quick way to just

cancel all tasks

...

Hey, is there a way for a flow to depend on another flow, with a different schedule. As in, wait execution until the first flow has completed (and vice versa)? I’ve seen running dependent flows topic, but that is constructing a higher-level flow. Where I want to have different schedules. Reason I need this: I sync several databases to BQ every 20 minutes, but on another schedule (once per day), I actually process the data in BQ. The data processing can’t run while sync are still running (due inconsistent data). So I have two crons, on different schedules, on which I want to create a dependency. Basically I want to hold a lock / semaphore, such that a flow waits until another flow releases it. Are there any suggestions to achieve this, preferably with core/server version? I’m not sure, but maybe what I’m looking for is described here in the cloud docs? And might it be implemented in this PR? Or do these tags behave differently?

Hi! is there a way to schedule a flow depending on a condition?

hi - would you all consider adding the project name to the cloudhook payload? I'm using cloudhooks to do some monitoring and it would be helpful to have the project name because we use projects to segment uat vs prod flows, as well as by teams within our org. If you're open to this, should I create an issue on the github repo?

Hey I have installed prefect server using k8s helm chart https://github.com/PrefectHQ/server/tree/helm-chart/helm/prefect-server and the services has running and how i can create prefect tenant and running k8s agent? Thank you

Hi, I need to run a fargate task in aws. I have followed this tutorial (https://docs.prefect.io/orchestration/execution/fargate_task_environment.html#examples) When I run this flow via flow.run(), the stdout print "Flow run SUCCESS: all reference tasks succeeded", the task does not run. What am I supposed to execute this flow?

Is there a way to pull the prefect client without the rest of prefect core? Think tehre may be only the one package on pypi

it seems like there may be some conflicts with cloudpickle when installing the prefect package with pipenv

- prefect [required: ==0.11.5, installed: 0.11.5]
    - click [required: >=7.0,<8.0, installed: 7.1.2]
    - cloudpickle [required: >=0.6.0,<1.5, installed: 1.4.1]
    - croniter [required: >=0.3.24,<1.0, installed: 0.3.36]
      - natsort [required: Any, installed: 7.0.1]
      - python-dateutil [required: Any, installed: 2.8.1]
        - six [required: >=1.5, installed: 1.14.0]
    - dask [required: >=0.19.3,<3.0, installed: 2.30.0]
      - pyyaml [required: Any, installed: 5.3.1]
    - distributed [required: >=1.26.1,<3.0, installed: 2.30.1]
      - click [required: >=6.6, installed: 7.1.2]
      - cloudpickle [required: >=1.5.0, installed: 1.4.1]
      - contextvars [required: Any, installed: 2.4]
        - immutables [required: >=0.9, installed: 0.14]
      - dask [required: >=2.9.0, installed: 2.30.0]
        - pyyaml [required: Any, installed: 5.3.1]
      - msgpack [required: >=0.6.0, installed: 1.0.0]
      - psutil [required: >=5.0, installed: 5.7.3]
      - pyyaml [required: Any, installed: 5.3.1]
      - setuptools [required: Any, installed: 45.2.0]
      - sortedcontainers [required: !=2.0.1,!=2.0.0, installed: 2.3.0]
      - tblib [required: >=1.6.0, installed: 1.7.0]
      - toolz [required: >=0.8.2, installed: 0.11.1]
      - tornado [required: >=5, installed: 6.1]
      - zict [required: >=0.1.3, installed: 2.0.0]
        - heapdict [required: Any, installed: 1.0.1]
    - docker [required: >=3.4.1,<5.0, installed: 4.3.1]
      - requests [required: >=2.14.2,!=2.18.0, installed: 2.25.0]
        - certifi [required: >=2017.4.17, installed: 2020.4.5.1]
        - chardet [required: >=3.0.2,<4, installed: 3.0.4]
        - idna [required: >=2.5,<3, installed: 2.9]
        - urllib3 [required: >=1.21.1,<1.27, installed: 1.26.1]
      - six [required: >=1.4.0, installed: 1.14.0]
      - websocket-client [required: >=0.32.0, installed: 0.57.0]
        - six [required: Any, installed: 1.14.0]
    - marshmallow [required: >=3.0.0b19,<3.6.1, installed: 3.5.1]
    - marshmallow-oneofschema [required: >=2.0.0b2,<3.0, installed: 2.1.0]
      - marshmallow [required: >=3.0.0rc6,<4.0.0, installed: 3.5.1]
    - mypy-extensions [required: >=0.4.0,<1.0, installed: 0.4.3]
    - pendulum [required: >=2.0.4,<3.0, installed: 2.1.2]
      - python-dateutil [required: >=2.6,<3.0, installed: 2.8.1]
        - six [required: >=1.5, installed: 1.14.0]
      - pytzdata [required: >=2020.1, installed: 2020.1]
    - python-box [required: >=3.4.4,<5.0, installed: 4.2.3]
      - ruamel.yaml [required: Any, installed: 0.16.12]
        - ruamel.yaml.clib [required: >=0.1.2, installed: 0.2.2]
      - toml [required: Any, installed: 0.10.2]
    - python-dateutil [required: ~=2.7, installed: 2.8.1]
      - six [required: >=1.5, installed: 1.14.0]
    - python-slugify [required: >=1.2.6,<5.0, installed: 4.0.1]
      - text-unidecode [required: >=1.3, installed: 1.3]
    - pytz [required: >=2018.7, installed: 2019.3]
    - pyyaml [required: >=3.13,<5.4, installed: 5.3.1]
    - requests [required: >=2.20,<3.0, installed: 2.25.0]
      - certifi [required: >=2017.4.17, installed: 2020.4.5.1]
      - chardet [required: >=3.0.2,<4, installed: 3.0.4]
      - idna [required: >=2.5,<3, installed: 2.9]
      - urllib3 [required: >=1.21.1,<1.27, installed: 1.26.1]
    - tabulate [required: >=0.8.0,<1.0, installed: 0.8.7]
    - toml [required: >=0.9.4,<1.0, installed: 0.10.2]
    - urllib3 [required: >=1.24.3, installed: 1.26.1]

I think actually problem is that prefect lib is conflicting with its own subdeps

maybe need to pin distributed and the change to cloudpickle is recent?

Hey prefect community, we have a somewhat philosophical question and a question of our fundamental understanding of prefect and orchestration (using dask): How do you approach the infrastructure setup, i.e. selection of node size, number of workers etc.? In a nutshell: • EKS vs Fargate vs others • fewer big nodes with high number of workers vs. many smaller nodes with lower number of workers • how many workers per node in general • how do you optimize between tasks, workers and nodes? What better questions to ask? How do you think and iterate over the infrastructure, prefect, dask setup, and so on? 🙂

Hi all, when we regsiter a flow , always return Error while deploying flow: FileNotFoundError(2, "No such file or directory: 'prefect'").

is it possible to run an agent in the cloud and not in my local machine?

Hi guys! I'm new to Prefect and I'm wondering if it has some built-in task to interact with ASW EMR...

Long time listener, first time caller. Wondering if there are any best practices for teams using Prefect Cloud to ensure that their execution environment is up to date? Are there any known issues for when Prefect Cloud moves to a new release and the execution environment a flow is running in doesn't move along side? What is Prefect's philosophy on versioning/breaking changes?

With the schedule concept, can I also set a maximum deadline for which a missed flow will ran? I just tested something while my agent wasn’t running, and now I got 40ish jobs it still will run.

There are any resources/examples about a ETL pipeline for live data ingestion reading from a Queue and for each new event it creates a new flow that is executed by a central dask cluster with auto scaling? (actually this is a good use case for prefect? or should we stick with batch jobs?)

Hi Everyone, I have a flow that have mapped tasks that use DaskCloudProvider Environnement and i have set the number of workers to 4. For some reasons only one worker is created. Here is the code:

flow.environment = DaskCloudProviderEnvironment(
    provider_class=FargateCluster,
    # task_role_arn="arn:aws:iam::497427061914:role/ecsTaskRole",
    execution_role_arn="arn:aws:iam::497427061914:role/ecsTaskRole",
    n_workers=4,
    scheduler_cpu=256,
    scheduler_mem=512,
    worker_cpu=512,
    worker_mem=1024,
    labels=["fargate"]
)