Hi! I’ve been setting common parameter for all the task such as

log_stdout=True, name="<task name>", state_handlers=[state_handler])

and wanted to have common parameter setted up to just call them. Is there any sample where such thing was done? or is there any way that I can set s common parameter that are being used in my company flows?

I find it a bit clunky that we have to either write all our code in one giant file, or rebuild + upload a docker image every time any of the other files change. Is Orion going to have a cleaner way to structure flows in multiple files?

Hi all! I have a Kubernetes cluster via AKS that’s running the Kubernetes agent. I’m trying to use the cluster autoscaler on node pools to scale the cluster depending on the cpu and memory requests of the prefect flows. We’re using prefect cloud for orchestration. The jobs are being submitted fine but they don’t seem to be consistently triggering scale up. Sometimes the exact same flow, with the same resource requests, will trigger the autoscaler, sometimes it won’t. When I take a look at the pods, the ones that run successfully and the ones that sit in pending have the same resource requests, etc. so they’re definitely feeding through ok to Kubernetes. We’re not maxing out the node limits of the pools (i.e. this still a problem scaling from 0->1) and the resources of the nodes (cpu/memory) are sufficient to run the flows. I’m at a loss for other reasons why flow runs that appear identical would sometimes work and sometimes not. I wanted to check if anyone had experienced this or had any ideas of where to look next?

Hi I've been playing around with prefect. Firstly, I think it is a really great tool!

I have a couple of question I hope you can help clarify my understanding.

Initially I had my example flow running in a docker container (on my local machine) through thr use of the

DockerRun

class and using

LocalStorage

. I had this running on a DockerAgent just fine.

I then changed the storage to

DockerStorage

, which again I got to work. I was just wondering though, does the use of the

DockerRun

class do anything when I am using

DockerStorage

? At first glance, it would seem not as the use of

DockerStorage

would imply the run would run on docker anyway. Am I missing something?

Also, it is currently not clear to me the difference between putting a selection of jobs in one flow as separate tasks vs having a selection of flows (of individual tasks) that have the dependencies set between them. In particular, I am interested in the execution resources for the different jobs.

I currently have a selection of jobs that run on kubernetes that I would like to migraate to prefect. I am trying to understand how I would make the decision of which approach to take. Is a flow on prefect deployed to a kubernetes pod and all the tasks deployed on that pod? If so and the current tasks have different configurations to run on kubernetes, would it be best to split the existing jobs into different flows?

Can I call

subflows

in the

parent flow

?

Hi, folks. I got 3 virtual machine(VM1, VM2, VM3) with access via corporative VPN. 1. Can I run Prefect Cloud on VM1? 2. How can I run agents on VM2&VM3 for Prefect server on VM1.

Hi, is there a way to make schedule to run one after one? like, if the first job didn’t finish, the second one has to wait.

Hey guys, I'm running into an issue with registering a flow that uses CronClock with parameter default inputs. On v. 0.15.5

ValueError: Flow could not be deserialized successfully. Error was: ValidationError({'schedule': {'clocks': {0: {'parameter_defaults': defaultdict(<class 'dict'>, {'accnt': {'value': ['Field may not be null.']}, ....

std_inputs = { 'report':False,
               'accnt':               None,  # string - delimited list
               'strategy':            None,  # string - delimited list
               'institutionalAdvisor':None,  # string - delimited list
               'endDate':             None,  # string - 'YYYY-MM-DD'
               'shadow':              None,  # SPY, treasury
               'benchmark':           None,  # SPY, PUT, BXM
               'fees':                False,  # Bool
               'output':              'daily'  # monthly, daily
               }

with Flow('Update Performance tables',
        schedule = Schedule(
                clocks = [
                      CronClock("7 6 * * 1-5",
                                start_date=pendulum.now("America/Chicago"),
                                parameter_defaults = std_inputs)]...

Any help would be greatly appreciated!

Hi, I have some parameters in the host_config for DockerRun that are in the child flows that I want to be able to edit/set when I go to run the parent flow. Is this currently possible in Prefect? • Is there a way for me to be able to define/set the host_config for the child process in the parent process?

Hi, I have a flow that was registered with a local Prefect server. If I invoke

flow.run()

, instead of it running on the registered server with the local agent it is running on my notebook. I could not find on the docs how to trigger the execution of the registered flow using the local server. Could you point me to an example or docs (which I likely missed)? What I am looking for is a way to execute the registered flow on my server via Python.

I have a flow that has 4 tasks, first being a parameter, second returning an intermediary result, third using that result to get some more intermediary results < fails here, fourth using all these intermediary results to run a query against my database. In testing I caused some failures on the 3rd step, and when trying to restart the failed job the intermediary results from the second step are set as None

Hi all, does anyone have experience importing and calling scrapy API scripts in Prefect? I am running into a

KeyError: 'Spider not found: 'scraper'

when I try to run debug my program. However when I try to define the function the task is calling manually, it works as intended.

Hi All, We're starting to onboard more and more people from our organization onto Prefect, and as we do so, one of the things we're noticing is that the flows that many people are authoring have the same handful of Python library dependencies – maybe 4 or 5 of the same libraries (e.g.

numpy

,

pandas

,

snowflake-connector-python

). Then our users will go and write a handful of their own custom Python classes and modules that they need for their flow. In order to make these custom modules available for use in their flows, people have been creating slight variations of the same docker image that have that same set of python packages installed in it, and then just

COPY

their project's code into the image – at that point, their RunConfig image has everything they need in it to run their flow. One of the issues I'm foreseeing with this approach is that it's going to lead to a lot of image bloat in terms of the number of images we'll have in use across our flows – images whose Dockerfiles might be found across several different repositories – so we'll be maintaining a lot of images that hardly differ from one another save for a handful of custom Python modules that people copy into them. I'm trying to see if there's an approach that avoids this – or at least avoids it in a way that has a favorable tradeoff. Maybe instead of these custom modules needing to be available at registration/build time, they can simply be retrieved at runtime from S3, for example? If that were possible, the management overhead now moves to S3 rather than our image repository, but I think that's easier to deal with; plus many of our users who need/want to build these flows don't necessarily want to be in the business in building and managing Docker images.

Hi guys, I am using prefect scheduler, for flow to repeat. Unfortunatelly, our task last longer than scheduler and that means that some things messes up. Is it possibility to set scheduler after previous successfull flow rerun same flow again and again.

Hi all, I'm getting an error in my Prefect run saying a column in my dbt model doesn't exist, but when I run the model locally everything works fine. I recently updated the Docker container so the code matches. Any ideas why I could be getting this?

Anyone else getting an issue like this with their Fivetran connectors in Prefect? `

Error during execution of task: HTTPError('404 Client Error: Not Found for url: <https://api.fivetran.com/v1/connectors/%0Alethal_conservation>')

Hello everyone! First time user here but excited to get going. I'm installing into kubernetes via helm and I'm not able to get the chart to create a tenant. It looks like its expecting apollo on localhost:4200 but its running in a different pod so I keep getting connection issues. Thoughts?

Hi! I’ve been using prefect for a pandas-heavy pipeline and I started to write some very redundant code. I can’t help but think there’s a nicer way than what I’m doing to achieve the desired result. I have a couple decorators I’m inserting into a function inside the task, then calling that function at the end. It works, but was wondering if I could extend the Task class or create one decorator with all the task calling stuff. My brain feels like a pretzel, is there any python superhero who’s been through this?

Hi everyone, I'm trying to setup prefect to use external db on aws rds and I verify that I can connect to it my rds with my user and password Here is my helm:

NAMESPACE=prefect-server
VERSION=2021.09.02

kubectl create namespace $NAMESPACE
helm install -n $NAMESPACE $NAMESPACE \
--version $VERSION \
--values ./helm/values.yaml \
prefecthq/prefect-server

Here is my helm

values.yml

postgresql:
  postgresqlDatabase: postgres
  postgresqlUsername: postgres
  existingSecret: [secret]
  servicePort: 5432
  externalHostname: "x.x.x.x"
  useSubChart: false

Is there anything else I'm missing to set in my value.yml? much appreciate it!

Hi, I have a problem with prefect core. I dynamically create a cron clock with different parameters and add them to the schedule. When I try to run them in one flow, then if the execution time overlaps, then only the first is executed. I thought it was solved https://github.com/PrefectHQ/prefect/pull/3394, but i have same problem.

def gen_prefect_cronclock(config, report_conf_path):
    clocks = []
    for report, cron in get_all_reports_cron_dict(report_conf_path).items():
        for tz, ppk_name_list in get_tz_ppk_dict(config).items():
            clocks.append(CronClock(cron,
                                    start_date=pendulum.datetime(
                                        1970, 1, 1, tz=tz),
                                    parameter_defaults={
                                        'report_path': report, 'ppk_name_list': ppk_name_list}
                                    )
                          )
    return clocks

run flow

with Flow("main-report-flow", main_schedule) as main_flow:
    conf = Parameter('config', default=config)
    report_config_path = Parameter('report_config_path',
                                   default=REPORT_CONF_PATH)
    report_path = Parameter('report_path', default="")
    ppk_name_list = Parameter('ppk_name_list', default=[])
    tasck_execute_send_report.map(unmapped(conf),
                                  unmapped(report_config_path),
                                  unmapped(report_path),
                                  ppk_name_list)

I have an interesting issue that I am running into. I have a flow that looks roughly like:

from settings import JOB_QUEUE, JOB_DEF

@task
def trigger_batch_job(input):
    ...
    return BatchSubmit(
        job_name=job_name,
        job_definition=JOB_DEF,
        job_queue=JOB_QUEUE,
        batch_kwargs=...,
    ).run()


with Flow(...):
    inputs = some_task()
    wait_res = AWSClientWait(
        client='batch',
        waiter_name='JobComplete',
    )(
        waiter_kwargs={
            'jobs': [
                trigger_batch_job(inputs),
            ],
            'WaiterConfig': {
                'Delay': 10,
                'MaxAttempts': 10000,
            },
        },
    )

And a settings.py file that looks like:

# settings.py
import os

JOB_DEF = os.environ.get(
    'PREFECT_FLOWS_JOB_DEF',
    'job-latest',
)

JOB_QUEUE = os.environ.get(
    'PREFECT_FLOWS_JOB_QUEUE',
    'job-gpu-dev',
)

The idea here is to have dynamic values for these settings based on what environment this is running in (we have dev/test/prod AWS accounts)

I ran a job in our test environment, and there was an error trying to submit the job to the dev queue (i.e. the default value here) I 2x checked and the environment variable is set on the docker image used to run the task

is it possible that this value is being predefined when the flow is built/registered and thus not using the most up to date value (the one at flow execution time)?

Is there a SLURM executor?

I've been trying to figure out how to properly expose the Dask Dashboard so it can be accessed externally outside of a flow's Docker Container on a flow run w/ DockerRun and DaskCluster Executor (ephemeral Local Dask Cluster), and the initial path I went down wasn't successful, so looking for some advice. Right now I have the Flow Runs running with containers on a user-defined Docker bridge network. Currently, can't reach the Dask Dashboard, which I assume is because its port (8787) isn't accessible outside the container. So, I figured the most straightforward way would be to enable port forwarding for the 8787 port (which Dask Dashboard runs on by default) for the containers. I was planning on adding in that extra config for port-forwarding with the

host_config

arg of

DockerRun

, however looking at the Docker Python SDK docs, it appears the ports for forwarding have to additionally be passed into the

create_container()

function (https://docker-py.readthedocs.io/en/stable/api.html#docker.api.container.ContainerApiMixin.create_container), which looking at the prefect source code, is not an additional arg I can pass into the run configs of

DockerRun

to thus be sent into

create_container()

. So, doesn't seem possible. This feels like something pretty common and doable, and I'm guessing I'm just looking at it the wrong direction. Has anyone accomplish this before and have any suggestions? Thanks!