https://prefect.io logo
Join the conversationJoin Slack
Channels
announcements
ask-marvin
best-practices-coordination-plane
data-ecosystem
data-tricks-and-tips
events
find-a-prefect-job
geo-australia
geo-bay-area
geo-berlin
geo-boston
geo-chicago
geo-colorado
geo-dc
geo-israel
geo-japan
geo-london
geo-nyc
geo-seattle
geo-texas
gratitude
introductions
marvin-in-the-wild
prefect-ai
prefect-aws
prefect-azure
prefect-cloud
prefect-community
prefect-contributors
prefect-dbt
prefect-docker
prefect-gcp
prefect-getting-started
prefect-integrations
prefect-kubernetes
prefect-recipes
prefect-server
prefect-ui
random
show-us-what-you-got
Powered by Linen
prefect-community
  • c

    Christian Nuss

    05/02/2022, 7:39 PM
    question! what has higher precedence? PREFECT__CONTEXT__SECRETS__AWS_CREDENTIALS or Items 3-8 from https://boto3.amazonaws.com/v1/documentation/api/latest/guide/credentials.html#configuring-credentials
    k
    • 2
    • 4
  • c

    Chris Reuter

    05/02/2022, 8:10 PM
    Quick question here - we're looking at doing a livestream with Hex. Anyone here using Hex in combination with Prefect in some way (maybe it's just part of your stack)?
    đź““ 1
    n
    • 2
    • 1
  • z

    Zach Schumacher

    05/02/2022, 9:18 PM
    hi - we started seeing a new log from prefect manifesting in some failed k8s jobs. Any idea what could be going on here?
    k
    g
    a
    • 4
    • 8
  • m

    Matt Alhonte

    05/02/2022, 9:56 PM
    Where in
    ECSRun
    would I put Dask security args? Like
    tls_ca_file
    and stuff?
    k
    • 2
    • 7
  • a

    Anthony L

    05/02/2022, 10:02 PM
    Hello team! I'm looking use Prefect at my company. While testing, I noticed that
    random
    generates the same values between runs. I'm using Prefect Cloud 1.0 and running a local agent (with Prefect 1.2.1 library installed). Here's my flow:
    import prefect
    from prefect import task, Flow
    from random import random
    
    @task
    def task1():
        logger = prefect.context.get("logger")
        random_val = random()
        <http://logger.info|logger.info>(f"{random_val=}")
        if random_val < 0.5:
            raise Exception()
    
    
    with Flow("random-bug-flow") as flow:
        task1()
    
    flow.register(project_name="ant-flows")
    I manually scheduled a run of this flow in Prefect Cloud and I see the random value in the logs. However, when I schedule another run (and any run thereafter), it generates the same "random" value. Is this expected?
    k
    a
    • 3
    • 12
  • g

    Gaurav

    05/02/2022, 10:52 PM
    Hello Prefect team! Hope you all are doing well 🙂 We are currently using Prefect 1.0 to launch tasks on AWS ECS using a local agent installed on an EC2 machine. The reason we are using a local agent is because we plan to do the same on other clouds (namely Azure and GCP) as well, and since they dont have their specific native agents (like ECSAgent for ECS), the local agent seemed more universal/generic approach for a multi cloud environment. We had some questions regarding the above however: 1. Using the local agent with the Prefect Cloud, is there a hard limitation or some throttling/delays in starting flows that can occur if we schedule a lot of tasks concurrently. For example running 250 flows (and the tasks therein) concurrently. Can a local agent support this? 2. I noticed Prefect 2.0 coming up soon, however upon reading the docs there, things like ECSAgent etc do not seem to exist there, and I was wondering what the migration process from 1.0 to 2.0 would look like. Basically - we would like to run our flows in a multi cloud setup (1 agent per cloud, tagged with respective labels). Thanks!
    k
    a
    • 3
    • 5
  • a

    Artem Vysotsky

    05/03/2022, 2:12 AM
    Hi community, I’m trying to find a documentation that will show an example (if possible) of programmatic cron job scheduling. I.e. I would imagine, that there will be a way to create a parametrized schedule via server api. Use case: Imagine a user interface where user could create a recurring job, specify job schedule and some additional parameters for the job to be executed (i.e. an ETL pipeline with sources and destinations configured via params)
    k
    • 2
    • 2
  • c

    Cole Murray

    05/03/2022, 5:19 AM
    Following up on https://prefect-community.slack.com/archives/CL09KU1K7/p1651543946765889 I wanted to confirm this was not an issue in Orion. In Orion, we've decoupled the one to many relationship between a Flow and Clocks (technically Schedule has the clocks) that was present with the introduction of a DeploymentSpec, which has a 1-1 with a schedule. Rather than iterating clocks here: https://github.com/PrefectHQ/server/blob/master/src/prefect_server/api/flows.py#L688 https://github.com/PrefectHQ/prefect/blob/cb0f80c6fc743cdae9f2af5b493ed454bc53d07c/src/prefect/schedules/schedules.py#L108 We are iterating deployments here: https://github.com/PrefectHQ/prefect/blob/orion/src/prefect/orion/services/scheduler.py#L83, which have a 1-1 between schedules. To solve the similar case of one flow, parameterized N many times & different schedules, we should be able to create N many DeploymentSpecs for the given flow without scaling issues, correct? cc: @Kevin Kho
    a
    • 2
    • 1
  • b

    Bernard Greyling

    05/03/2022, 5:45 AM
    Hello We've been using prefect internally for about two years now. Congrats on the cadence of innovation. I'ts been a pleasure using your tools. At this stage we have two environments setup, one for ETL runs and another for internal reporting mechanisms. We are on the standard plan and would like to expand the number of users. I've tried twice now to get access to the sales team via the main prefect portal. Haven't heard anything from you guys in over two weeks, Can someone please assist me in having the relevant conversation ?
    a
    • 2
    • 1
  • d

    davzucky

    05/03/2022, 6:25 AM
    I need a script that restart a lot of flow run. I'm able to get the flow runs, however I'm not sure if the mutation to restart them. We had about 600 failed run because of infra that I need to restart
    a
    • 2
    • 5
  • s

    Suresh R

    05/03/2022, 6:37 AM
    Hi! I am getting below error sometimes when i try to update a Prefect secret, i suspect multiple flows updating same secret at same time causing this issue, still i am not sure. Can someone help.
    prefect.exceptions.ClientError: [{'path': ['secret_value'], 'message': 'An unknown error occurred.', 'extensions': {'code': 'INTERNAL_SERVER_ERROR'}}]
    a
    k
    • 3
    • 4
  • d

    Domenico Di Gangi

    05/03/2022, 7:09 AM
    Hi all, I am using prefect 2.0 more and more these days and loving it! Is there a way to quickly see execution's time of a flow run from the UIs listings? I guess I could filter runs, but it would be nice to have a faster way to select a run based on its execution time. It seems that the visualization has enough space available in the runs' rows and maybe timestamp could easily fit on it:
    a
    m
    j
    • 4
    • 12
  • t

    Thomas Mignon

    05/03/2022, 7:29 AM
    Hi guys, I'm trying to pass from a flow of 6 task to 6 flow of 1 task with daskExecutor
    a
    m
    • 3
    • 31
  • s

    Sang Young Noh

    05/03/2022, 8:23 AM
    Hi I'm currently trying to do some deployment testing, and I'm getting the following error: Specification in 'newflow.py', line 13 failed validation! You have not configured default storage on the server or set a storage to use for this deployment but this deployment is using a Universal flow runner which requires remote storage. I've currently got a file called: from prefect import flow @flow def hello_world(name="world"): print(f"Hello {name}!") # Note: a deployed flow does not need a command to # explicitly run the flow. The API handles this for you. # hello_world() from prefect.deployments import DeploymentSpec DeploymentSpec( flow=hello_world, name="hello-world-daily", ) and the error comes out when I run the following: prefect deployment create newflow.py
    âś… 1
    a
    v
    • 3
    • 102
  • a

    Andreas Nord

    05/03/2022, 8:54 AM
    After upgrading Prefect I get this warning on tasks with retry settings:
    UserWarning: Task <Task: task_name> has retry settings but some upstream dependencies do not have result types. Cloud features such as input caching and resuming task runs from failure may not work properly.
    The (automatic) retrying seems to be working fine, not sure what this warning refers to. I am aware that I cannot resume task run from failure, but not sure why that is related to retry settings
    a
    • 2
    • 1
  • r

    Ron Levi

    05/03/2022, 10:18 AM
    I was wondering how can one access/modify Prefect Context variables at Task/Flow run on Prefect 2.0 (similar to how it was done in Porter 1 -> https://docs.prefect.io/core/concepts/execution.html#prefect-supplied-context)
    a
    • 2
    • 2
  • a

    Amruth VVKP

    05/03/2022, 10:45 AM
    I am getting a very wired error on Prefect Orion 2.0b3 when I am attempting to run a Prefect Shell task Here's how my flow is structured - Scenario 1 - Primary Flow: Flow is not a async flow. Uses sequential task runner, a bunch of processing tasks. The final task is to build a CLI and trigger a shell task that runs a particular test, makes few API calls, etc.
    @flow(name='primary_flow', version='1', task_runner=SequentialTaskRunner())
    def primary_flow(**kwargs):
        ... # perform data processing tasks, all tasks are non-async
        for cli_command in parsed.result():
            shell_run_command(cli_command) # I've also tried using asyncio.run for running the shell task
    Scenario 2 - Primary Flow: Uses sequential task runner, a bunch of processing tasks. The final task to build and call a CLI is done through another async subflow
    @flow(name='primary_flow', version='1', task_runner=SequentialTaskRunner())
    def primary_flow(**kwargs):
        ... # perform data processing tasks, all tasks are non-async
        for _ in parsed.result():
            cli_flow(_) # I've also tried using asyncio.run for running the shell task
    
    @flow(name='cli_flow', version='1', task_runner=SequentialTaskRunner())
    async def cli_flow(parsed):
        for _ in parsed:
            ... # Build CLI
            await shell_run_command(cli_command)
    I am getting this exception thrown (attached screenshot) and the CLI task is not getting started/executed. Can someone help me out on this one please (I am terrible with asnyc processes)?
    :discourse: 1
    a
    • 2
    • 13
  • m

    Mars

    05/03/2022, 2:29 PM
    Hi all, I’m having trouble diagnosing a GitHub storage problem. I’ve created a trivial testing flow similar to the example script-based workflow for GitHub. I’ve deployed a k8s agent using
    prefect k8s agent install
    . I’ve uploaded my flow to a private GitHub repo and registered it with Prefect. And I’ve added a Cloud Secret called
    GITHUB_ACCESS_TOKEN
    that holds a valid GitHub personal access token. When I run my flow the agent’s GitHub storage gives me an
    UnknownObjectException(404, 'Not Found')
    error. If I change the flow to use a different Cloud Secret key for the PAT, such as
    access_token_secret='MYKEY'
    , then the agent tells me
    ValueError('Local Secret "MYKEY" was not found.')
    . How can I introspect the kubernetes agent to verify that the GitHub PAT secret is being loaded from Prefect Cloud correctly?
    k
    • 2
    • 62
  • j

    Jai P

    05/03/2022, 3:09 PM
    đź‘‹ hullo folks! have a general question about flows vs. tasks vs. just regular functions in prefect 2.0. are there any high level guidelines on where we may want to use each? i'll try to describe a scenario in the thread
    :discourse: 1
    đź‘‹ 2
    m
    a
    • 3
    • 12
  • j

    Jakub Cieslik

    05/03/2022, 4:54 PM
    Hi, i have a quick question about referencing a past flow run in the current-flow. Lets say i'm processing files in one-run of the flow, the next time i'm executing (the flow) it i want to only process the new filesnot yet processed, so i could process only those that were created after the last successful run (maybe add some overlap just in case) Pretty sure doing it times based is not the best solution but it would work for my case. For simplicity i don't want to keep a state of already processed files. What would be a common solution to this problem?
    k
    • 2
    • 7
  • g

    Gowtham Manne

    05/03/2022, 5:22 PM
    Hi All, General Question - Can we deploy prefect on open shift ? Is there a doc for this ? Appreciate your help!
    :discourse: 1
    k
    a
    d
    • 4
    • 16
  • d

    David Beck

    05/03/2022, 5:29 PM
    Hi all! I'm looking for some guidance on handling logging of context attributes. To best organize our events/metrics within out internal systems, I'm looking to prefix all Prefect context attributes with
    prefect.
    , e.g.
    prefect.flow_name
    . Given the documentations cautioning of renaming context attributes, I'm trying to brainstorm the simplest and most effective way of doing this across all our flows. Do you have suggestions?
    k
    • 2
    • 8
  • a

    Andrew Lawlor

    05/03/2022, 6:25 PM
    i ran a flow of flows process that was supposed to kick off 12,000 flows on GKE. it started 10,000 flows, and then failed with the message
    Pod prefect-job-5e3af599-tl2xs failed. No container statuses found for pod
    where can i look for a more detailed message? any idea what actually caused it to fail? also, of those jobs, all but one passed. the one that failed had the same message (no container statuses found).
    :discourse: 1
    k
    a
    • 3
    • 23
  • a

    Artem Vysotsky

    05/03/2022, 6:55 PM
    Okay, prefect is officially awesome. I just launched the server locally and realized i didn’t see the whole picture from the docs. Thanks for building such an amazing product.
    :thank-you: 4
    :marvin: 6
    k
    j
    • 3
    • 2
  • c

    chicago-joe

    05/03/2022, 7:34 PM
    Every few days, some of our DockerRun scripts fail with a weird 500 Internal Server Error: ("error while mounting volume '/var/lib/docker/volumes/sraprod67-output/_data': failed to mount local volume: mount but most flow runs are successful, with no mounting issues. Has anyone else experienced this?
    k
    • 2
    • 3
  • b

    Bradley Hurley

    05/03/2022, 8:34 PM
    Hi Prefect Experts - Im looking for a little guidance on using the
    ExecuteNotebook
    Task.
    k
    • 2
    • 16
  • a

    Andrew Lawlor

    05/03/2022, 8:41 PM
    im looking to run a flow on a schedule of every day except for every other monday (on those days i want to run a different flow). what is the best way to do that via a schedule?
    k
    a
    • 3
    • 2
  • m

    Matthew Roeschke

    05/03/2022, 9:07 PM
    I’m looking to set up an automated way to prune old/test flows in a project. I found the GraphQL endpoint to remove flows, but to determine old flows vs current flows I was thinking to: 1. Pipe the output of
    prefect register -p project
    to get the names of the current flows that would have been registered 2. Get the IDs of the flows through the GraphQL API 3. Delete flow IDs that that do not have names from step 1 Is there a more sensible way to get the “current flows” in a project beside the
    prefect register
    trick?
    k
    • 2
    • 8
  • m

    Matt Alhonte

    05/03/2022, 9:35 PM
    Hrm, looks like the Client is dying shortly after launching the Cluster? It runs a few tasks, then right before mapping a big-ish one, the Client dies (cloudwatch logs just say
    Killed
    ) and the mapping stays in a
    mapped
    state
    a
    k
    • 3
    • 19
  • c

    Cole Murray

    05/04/2022, 5:45 AM
    Hey Prefect team, Is there appetite for adding a delete to the CLI deployments command for orion? Recently ran into a case where I need to delete some deployments, but didn't find it available via CLI It looks like the change would require two changes: Add Api method for delete_deployemnt https://github.com/PrefectHQ/prefect/blob/d591e730ee160878eb21755464a234cef1f16b47/src/prefect/client.py#L168 Add Delete command to cli calling above delete_deployment: https://github.com/PrefectHQ/prefect/blob/d591e730ee160878eb21755464a234cef1f16b47/src/prefect/cli/deployment.py
    🎉 1
    đź’Ż 1
    a
    • 2
    • 2
Powered by Linen
Title
c

Cole Murray

05/04/2022, 5:45 AM
Hey Prefect team, Is there appetite for adding a delete to the CLI deployments command for orion? Recently ran into a case where I need to delete some deployments, but didn't find it available via CLI It looks like the change would require two changes: Add Api method for delete_deployemnt https://github.com/PrefectHQ/prefect/blob/d591e730ee160878eb21755464a234cef1f16b47/src/prefect/client.py#L168 Add Delete command to cli calling above delete_deployment: https://github.com/PrefectHQ/prefect/blob/d591e730ee160878eb21755464a234cef1f16b47/src/prefect/cli/deployment.py
🎉 1
đź’Ż 1
Delivery: 📬 https://github.com/PrefectHQ/prefect/pull/5750/files
:marvin: 1
a

Anna Geller

05/04/2022, 9:30 AM
Thanks for becoming a Prefect 2.0 contributor 🎉 I added one tiny comment
âś… 1
View count: 4