https://prefect.io logo
Docs
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
  • p

    Peter

    06/23/2020, 12:01 AM
    We are working on integrating prefect with our existing applications and we've run into an issue with logging. Effectively, it looks like that a Prefect flow disables existing loggers when you call
    flow.run()
    . I've read through the documentation on logging which recommends you use prefect's utility function
    get_logger()
    to hook into a flow's logs, which works fine for creating a new application but would be a ton of work for us to change our existing applications that use python's standard logging practice - essentially defining
    logger = logging.getLogger(__name___)_
    at the top of the module. Unfortunately, when we call
    flow.run()
    with tasks that invoke our existing application that uses standard logging it looks like prefect disables those logs. Is there any way to tell prefect to log from existing applications that use python's standard
    logging.getLogger
    instead of prefect's
    get_logger
    utility?
  • p

    Peter

    06/23/2020, 12:01 AM
    Also here's a toy script that illustrates my issue:
    logger_issue.py
    m
    c
    • 3
    • 5
  • s

    Sandeep Aggarwal

    06/23/2020, 1:36 AM
    Sharing on channel as well if anyone else would like to chime in.
    n
    • 2
    • 2
  • k

    Kai Weber

    06/23/2020, 5:59 AM
    Hi can anybody give me a hint what the GraphQL query in HASURA to start my flow "Hello World" has to look like? This query shows my flows: query MyQuery { flow(distinct_on: name) { name id description } }
    n
    • 2
    • 12
  • s

    Simone Cittadini

    06/23/2020, 7:11 AM
    Hi, I have a strange error in a docker-swarm deploy : if I run "curl http://apollo:4200/graphql" from the ui container I get ( I think correctly ) "GET query missing." but the UI itself under API status says "Couldn't connect http://apollo:4200/graphql/"
    n
    • 2
    • 4
  • s

    Simone Cittadini

    06/23/2020, 8:02 AM
    On a side note, I've noticed that if you call register on the same flow ( no code changes ) multiple times it creates multiple versions, is it considered a bug or that is how it works ? I can't let anyone register whatever they want from the laptop, so in my plan this is how it will work: • code is made in a branch of a common "all flows" repository • once approved is merged and the pipeline deploys the package in the agent, which proceeds to register everything If I dumbly register everything all the n flow already existing and untouched will bump their revision, not optimal ( not a problem if this is just how it works, I'll make the scan of code less dumb with some hash checking, but of course if it's just a matter of waiting a fix I'll gladly just wait 😁 )
    j
    j
    • 3
    • 2
  • m

    Matias Godoy

    06/23/2020, 12:52 PM
    Hi guys! I'm trying to run a flow using the GraphQL API in Prefect Cloud by executing:
    mutation {
      create_flow_run(input: {
        flow_run_name: "Test"
        version_group_id: "81261519-e91c-4fe3-bf85-10cc3a2d5016"
      }) {
        id
      }
    }
    But I get the following error:
    {
      "graphQLErrors": [
        {
          "path": [
            "create_flow_run"
          ],
          "message": "Version group 81261519-e91c-4fe3-bf85-10cc3a2d5016 has no unarchived flows.",
          "extensions": {
            "code": "INTERNAL_SERVER_ERROR"
          }
        }
      ],
      "networkError": null,
      "message": "GraphQL error: Version group 81261519-e91c-4fe3-bf85-10cc3a2d5016 has no unarchived flows."
    }
    I'm sure the group ID is correct, and that there is a version of a test flow I created. I can even run it manually. What am I doing wrong? Thanks!
    z
    • 2
    • 5
  • h

    Howard Cornwell

    06/23/2020, 1:19 PM
    Can anyone explain to me why this raises an error?
    import prefect
    import sqlalchemy as sql
    
    
    engine = sql.create_engine("<postgresql://tmp:tmp@localhost:5432/tmp>")
    tmp_schema = sql.MetaData()
    table = sql.Table("tmp", tmp_schema, sql.Column("tmp", sql.Integer))
    
    
    @prefect.task
    def create_table():
        tmp_schema.create_all(engine, tables=[table])
    
    
    with prefect.Flow("tmp") as flow:
        create_table()
    
    
    flow.storage = prefect.environments.storage.Docker(
        registry_url="localhost:5000",
        python_dependencies=["sqlalchemy"]
    )
    flow.register()
    Raises
    Traceback (most recent call last):
      File "issue.py", line 23, in <module>
        flow.register()
      File "/Users/howardcornwell/files/dev/misc/prefect-pickle-issue/.venv/lib/python3.8/site-packages/prefect/core/flow.py", line 1437, in register
        registered_flow = client.register(
      File "/Users/howardcornwell/files/dev/misc/prefect-pickle-issue/.venv/lib/python3.8/site-packages/prefect/client/client.py", line 649, in register
        serialized_flow = flow.serialize(build=build)  # type: Any
      File "/Users/howardcornwell/files/dev/misc/prefect-pickle-issue/.venv/lib/python3.8/site-packages/prefect/core/flow.py", line 1299, in serialize
        storage = self.storage.build()  # type: Optional[Storage]
      File "/Users/howardcornwell/files/dev/misc/prefect-pickle-issue/.venv/lib/python3.8/site-packages/prefect/environments/storage/docker.py", line 293, in build
        self._build_image(push=push)
      File "/Users/howardcornwell/files/dev/misc/prefect-pickle-issue/.venv/lib/python3.8/site-packages/prefect/environments/storage/docker.py", line 329, in _build_image
        dockerfile_path = self.create_dockerfile_object(directory=tempdir)
      File "/Users/howardcornwell/files/dev/misc/prefect-pickle-issue/.venv/lib/python3.8/site-packages/prefect/environments/storage/docker.py", line 436, in create_dockerfile_object
        cloudpickle.dump(self._flows[flow_name], f)
      File "/Users/howardcornwell/files/dev/misc/prefect-pickle-issue/.venv/lib/python3.8/site-packages/cloudpickle/cloudpickle_fast.py", line 48, in dump
        CloudPickler(file, protocol=protocol, buffer_callback=buffer_callback).dump(obj)
      File "/Users/howardcornwell/files/dev/misc/prefect-pickle-issue/.venv/lib/python3.8/site-packages/cloudpickle/cloudpickle_fast.py", line 548, in dump
        return Pickler.dump(self, obj)
    TypeError: cannot pickle '_thread._local' object
    z
    j
    • 3
    • 5
  • m

    Miguel

    06/23/2020, 1:44 PM
    Hi everyone I'm migrating an airflow dag to prefect and prefect server and trying to figure out how to backfill this flow. I've created a function that uses the
    client.create_flow_run
    to generate all the necessary runs with the correct context but, at least using the local agent, this creates way too many processes (one for each run, I guess). Is this this right approach or is there a way to limit the concurrency / number of processes spawned?
    z
    n
    • 3
    • 4
  • j

    james.lamb

    06/23/2020, 3:24 PM
    Good morning from Chicago! I'd like to get confirmation on something I think I've discovered. Is this true?
    If you use
    KubernetesJobEnvironment
    as the execution environment for a flow, any
    executor
    you pass will be ignored and only the default executor will be used
    j
    • 2
    • 15
  • j

    Jacques

    06/23/2020, 3:51 PM
    I'm trying to get to the bottom of a weird error, hoping this rings a bell for someone - I'm making a boto call in one of my tasks, and if the task fails I set it to retry with a
    max_retries
    parameter. I'm catching the error from boto and then logging the error immediately before doing a
    raise signals.FAIL()
    to trigger the retry mechanism. When the boto call fails (it does this once or twice a day - unpredictably) the error is caught, logs show the task is set to
    Retrying
    , and downstream tasks are set to
    Pending
    . All looks good until the flow is scheduled to run again, then I get a python stack overflow as some object is being pickled (I think - seeing looped calls to bits like
    File "/var/lang/lib/python3.7/pickle.py", line 662 in save_reduce
    in the stack trace) directly after the
    Beginning Flow run
    message. I'm using
    DaskExecutor
    if that matters.
    j
    • 2
    • 14
  • j

    jeff sadler

    06/23/2020, 5:04 PM
    Hi, I've been trying to scale out my flow with a Dask Executor object. I worked through some initial pickling errors, but now I'm getting an error with no explanation. Is there a way to see where my flow if failing and get some more diagnostic info? Here's all that I see:
    [2020-06-23 16:56:03] INFO - prefect.FlowRunner | Beginning Flow run for 'run_model'
    [2020-06-23 16:56:03] INFO - prefect.FlowRunner | Starting flow run.
    distributed.scheduler - INFO - Receive client connection: Client-6c8e597e-b572-11ea-ace9-ac1f6b4e4528
    distributed.core - INFO - Starting established connection
    distributed.scheduler - INFO - Remove client Client-6c8e597e-b572-11ea-ace9-ac1f6b4e4528
    distributed.scheduler - INFO - Remove client Client-6c8e597e-b572-11ea-ace9-ac1f6b4e4528
    distributed.scheduler - INFO - Close client connection: Client-6c8e597e-b572-11ea-ace9-ac1f6b4e4528
    [2020-06-23 16:56:05] INFO - prefect.FlowRunner | Flow run FAILED: some reference tasks failed.
    It's hard to know where to start with only
    Flow run FAILED: some reference tasks failed.
    z
    j
    • 3
    • 30
  • c

    Chris Hart

    06/23/2020, 8:34 PM
    has anyone used https://github.com/mara/mara-pipelines before and have any impressions?
    b
    • 2
    • 1
  • m

    Michael Hoss

    06/24/2020, 8:43 AM
    Dear Prefect community, can someone provide a direct comparison of Prefect and Luigi (https://github.com/spotify/luigi)? I have the impression that Prefect is superior in many aspects, but would like to be sure about that without spending much more time on trying out Luigi myself.
    a
    d
    s
    • 4
    • 4
  • p

    psimakis

    06/24/2020, 12:44 PM
    Hello Prefect Community, I face this issue. Have you any idea/advice? Thanks in advance!
    j
    • 2
    • 4
  • i

    itay livni

    06/24/2020, 1:39 PM
    Hi - Here is a link to a question I posted in the wrong channel. https://prefect-community.slack.com/archives/CL09KTZPX/p1592962127232500
    👍 1
  • m

    Marwan Sarieddine

    06/24/2020, 4:59 PM
    Hi folks, I am trying to programmatically “timeout a flow run” by marking it as failed using
    prefect.client.client.Client
    - looking at the docs I see:
    set_flow_run_state(
            self, flow_run_id: str, version: int, state: "prefect.engine.state.State"
        )
    where I can pass the
    flow_run_id
    and
    state
    , but I am trying to understand the intuition behind setting
    version
    Would I have to find the version of the task the flow is currently on and increment it ?
    c
    m
    • 3
    • 6
  • b

    Braun Reyes

    06/24/2020, 11:25 PM
    Is the environment metadata a new thing? Looks like this gets sent to cloud. Would be interesting to use this for kwargs fargate agentbcpuld leverage
    j
    • 2
    • 2
  • r

    Rafal

    06/25/2020, 7:59 AM
    Hello, any experience how to lunch containers in prefect? How to make this container accessible on localhost port? I mean running containers in workflow and make them accessible during pipeline execution
    d
    • 2
    • 5
  • h

    Howard Cornwell

    06/25/2020, 10:47 AM
    Has anyone experienced using prefect for lots (millions) of API requests? I initially created one mapped task for each request, but this caused issues with prefect server performance. I’ve batched the requests into ~5000 per child task, but I can only retry entire batches, not single requests. I’m on the verge of tracking the state of each request outside of Prefect, but wonder if anyone has experienced / overcome similar issues?
    d
    • 2
    • 4
  • e

    Eamon Keane

    06/25/2020, 11:00 AM
    I'm still a little unclear on the kubernetes functionality in prefect (after reading the docs). Can I launch a
    CreateNamespacedPod
    task and will the prefect agent monitor the launched pod's exit status (via kubernetes api) as an indicator of task success/failure?
    d
    • 2
    • 10
  • d

    Darragh

    06/25/2020, 11:36 AM
    Monitoring/Logging question - we’ve been seeing some odd behaviour that we’re still trying to classify, around parallell mapped tasks. We’ve rolled our own subflow semantics by taking a set of inputs [generally partitioned data files], and for numInputs we map over them and trigger a Flow with that input/file. This allows us to split out the work so we can have a bunch of Fargate nodes attack chunks of the data processing. Once these flows have been triggered, we have another mapped task that polls the flows for completion. By and large it works fine, but we see strange cases where some of the mapped tasks can do the following: • Submit and not start, so polling sits there forever waiting on it, or just fail because it never starts • Some mapped tasks, the pollers in particular, don’t actually get started for a few minutes after the others in the mapped input. For example -> mapped_lows = [flow_id_1, flow_id_2, flow_id_3] => poll_flow(mapped_flows) => poller_1, poller_2 start straightaway, poller_3 not triggered for several minutes with no indication as to why
    👀 1
    l
    • 2
    • 8
  • j

    josh

    06/25/2020, 1:58 PM
    Hey team, Prefect core version 
    0.12.1
     has been released! Here are a few notable highlights: 📁   File-based storage for flows that unlocks the ability for flows to “hot reload” 🚢   Flows stored in blob/bucket storage can be orchestrated with containerized agents/environments :dask:   Improved/streamlined environment and executor relationship 🏗️   Task slugs are now stable across rebuilds of the same flow Read the full changelog here: https://github.com/PrefectHQ/prefect/releases/tag/0.12.1 Special thanks to all of the contributors who have assisted in discussions, issues, and PRs that have contributed to this release!
    😍 7
    :dask: 1
    💯 6
    🚀 9
    👍 8
  • j

    Jeff Brainerd

    06/25/2020, 5:23 PM
    Hi team, a Prefect mystery (at least for me 😬) — I’m using Prefect Cloud and Prefect Core 0.11.4, also Fargate Agent and docker storage. My goal is to call the Prefect API from a state handler, so I can report flow stats to Slack like flow duration when a flow finishes. Problem is I am getting an unauthenticated error. I can see from
    agent.py
    that the container env should include the env var:
    "PREFECT__CLOUD__AUTH_TOKEN": config.cloud.agent.auth_token,
    So at this point I’m not sure if that value is not set, or it is set correctly but the agent token doesn’t have the correct auth. PS — I am open to doing this another way, such as a dedicated SlackTask, but the state handler seems somehow more semantically correct, and the CloudHook doesn’t seem to provide that kind of detailed info. (Sorry for the long post.) Thanks! 🙏
    c
    m
    • 3
    • 10
  • h

    Hannah Amundson

    06/25/2020, 8:25 PM
    Hello Prefect Community! I have a question about the KubernetesJobEnvironment. Is there a way to see what job environment has been stored/is being used? I would be fine with either something through the interactive API or getting into the docker image
    👀 1
    l
    • 2
    • 2
  • j

    Jeremiah

    06/25/2020, 9:58 PM
    👋 Hi everyone <!here>! A little less than a year ago, we created this Slack community just in case anyone would actually want to chat with us. The one commitment we made was to make this a welcoming place that would quickly result in an expert answer to any question. At some point earlier today, we welcomed the thousandth member of this community! That’s honestly mind-blowing to me, and from our entire team please know we’re very grateful for the interest and engagement of all of you. Over the last year, it’s been our pleasure to build relationships, develop features, discuss the future, and — just maybe once or twice — track down bugs with you. Thanks so much for being part of our community — I promise we’ve got plenty in store that you’re going to love. Happy engineering!
    💥 11
    🤘 4
    👏 4
    🍾 14
    :marvin: 13
    :prefect: 16
    ❤️ 24
  • a

    Alfie

    06/26/2020, 2:55 AM
    Hi Team,
  • a

    Alfie

    06/26/2020, 2:57 AM
    I’m new to Prefect, and want to know if Prefect exposes any REST API. I have the frontend UI already and want to figure out a way to talk with Prefect
    c
    • 2
    • 2
  • a

    Alfie

    06/26/2020, 2:57 AM
    Thanks.
  • k

    Kostas Chalikias

    06/26/2020, 8:38 AM
    Hello everyone, a few of us have noticed that when we load the Prefect front end of our cloud instance on a Chrome tab, do a few things then go about our day and return to said tab later, it's basically always gone blank and we need to close the tab and open another one to get the front end back. Is this just us?
    a
    n
    • 3
    • 7
Powered by Linen
Title
k

Kostas Chalikias

06/26/2020, 8:38 AM
Hello everyone, a few of us have noticed that when we load the Prefect front end of our cloud instance on a Chrome tab, do a few things then go about our day and return to said tab later, it's basically always gone blank and we need to close the tab and open another one to get the front end back. Is this just us?
a

Arsenii

06/26/2020, 9:04 AM
Same issue!
👍 2
k

Kostas Chalikias

06/26/2020, 4:53 PM
Any ideas on this?
n

nicholas

06/26/2020, 4:58 PM
Hi ya'll, is this the Prefect Core UI (Prefect Server) or Prefect Cloud?
k

Kostas Chalikias

06/26/2020, 5:31 PM
Prefect cloud for us
n

nicholas

06/26/2020, 5:42 PM
Got it, thanks for bringing that to our attention @Kostas Chalikias - I'll look into that to see what's up.
k

Kostas Chalikias

06/26/2020, 5:50 PM
Let me know if you need any diagnostic info from my side
n

nicholas

06/26/2020, 5:51 PM
If you could DM me your user email, that'd be perfect!
👍 1
View count: 1