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?

Also here's a toy script that illustrates my issue:

Sharing on channel as well if anyone else would like to chime in.

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 } }

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/"

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 😁 )

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!

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

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?

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

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.

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.

has anyone used https://github.com/mara/mara-pipelines before and have any impressions?

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.

Hello Prefect Community, I face this issue. Have you any idea/advice? Thanks in advance!

Hi - Here is a link to a question I posted in the wrong channel. https://prefect-community.slack.com/archives/CL09KTZPX/p1592962127232500

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 ?

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

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

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?

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?

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

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!

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! 🙏

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

👋 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!

Hi Team,

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

Thanks.