Is it possible to drop the index with the PandasSerializer with serialize_kwargs?

Where can I find

~/.prefect/config.toml

in Windows 10

For S3Result.write(), Returns: • `Result`: a new Result instance with the appropriately formatted S3 URI I wasn't able to return the formatted S3 URI, only the S3 Result itself. Any pointers? I'm doing:

data = results.write(df, **prefect.context)
print(data)

Hello everyone, for CI/CD purposes, we need to call prefect from Jenkins. There is no Restful API in prefect, so we need to use GraphQL. Is there any good example to share for Jenkins / Prefect integration?

What is the purpose of

PREFECT__CLOUD__AUTH_TOKEN

if the

prefect auth login

requires a token to authenticate? Are these incompatible options. I am trying to authenticate with Cloud backend during a CI deployment and have set the env var for the auth token, but my flow fails to register.

How can we configure prefect cloud to use Gitlab as Storage for a particular project rather defining that in each flow?

seeing this error on Prefect cloud when attempting to programatically start a flow run. Our authentication tokens have not changed so I'm wondering if something happend?

File "/usr/local/lib/python3.8/site-packages/prefect/client/client.py", line 978, in create_flow_run
    res = self.graphql(create_mutation, variables=dict(input=inputs))
  File "/usr/local/lib/python3.8/site-packages/prefect/client/client.py", line 275, in graphql
    result = <http://self.post|self.post>(
  File "/usr/local/lib/python3.8/site-packages/prefect/client/client.py", line 230, in post
    response = self._request(
  File "/usr/local/lib/python3.8/site-packages/prefect/client/client.py", line 373, in _request
    token = self.get_auth_token()
  File "/usr/local/lib/python3.8/site-packages/prefect/client/client.py", line 503, in get_auth_token
    self._refresh_access_token()
  File "/usr/local/lib/python3.8/site-packages/prefect/client/client.py", line 618, in _refresh_access_token
    payload = self.graphql(
  File "/usr/local/lib/python3.8/site-packages/prefect/client/client.py", line 294, in graphql
    raise ClientError(result["errors"])
prefect.utilities.exceptions.ClientError: [{'path': ['refresh_token'], 'message': 'Unauthorized', 'extensions': {'code': 'FORBIDDEN'}}]

So, at a new company and helping them set up their data infra! What would be the smoothest way to get up and running with Prefect Cloud on GCP and with a pretty small team? Running the flows on GKE?

I am getting an unexpected error when trying to use

apply_map

in conjunction with skipped tasks.

At least one upstream state has an unmappable result.

Hello there, I've noticed lately the prefect cloud UI refuses to load completely, here is what I see on my network tab today (yesterday the screen was completely blank but similar network activity). Also back when it used to work, if left alone in a tab for a few mins it would crash and I had to reload - I have reported this months ago as well

Hey, is it possible to have extra_loggers on a diffrent log_level than PREFECT__LOGGING__LEVEL? I´d like to have my custom logger on info level, and CloudTaskRunner and CloudFlowRunner logger on warning

Hi there ! I'd like to use

prefect.utilities.notifications.slack_notifier

behind a proxy. A search in the docs and in the slack community give nothing so before submitting a new issue I wanted to check if I was not missing something ! 🙂 Thanks

hey folks, high level question here. I'm building a system that will hit an SQS queue for a pub/sub thing capturing point in time updates to a dataset. The flow from the queue is not terribly big, but will have a few million or more dispatches during normal working hours. I'll want to hit the queue, parse the message (the payloads are complete messages with no external info necessary, and all strings/bytes), do minor transforms, and then send an update query to a relational DB with a soft real-time limit of around one second. Is there any resources as to the utility of prefect in this kind of use case? I'm interested in using a data pipeline (vs. a boutique app) in this context because there's a lot more pipelines I need to build, and would rather circle them all up in one spot. What say ye, people of #prefect-community?

What is the meaning of a ⚠️ on the Prefect Cloud status page at https://prefect.status.io/? (details in thread)

Hey everyone, just wanted to remind you guys what we’ll be having a livestream at 4 pm ET (2.5 hours from now) about orchestrating data science workflows on Prefect.

Join our Meetup event.

Is there a way to pin a project to specific agent only? E.g I have a group of downloading flows in project A, I wish all flows in project A only run on agent B where the cert/credentials are available

Is there a way to erase the by default host name label? I call the code below, however, my flow ends up with 2 label. I see an option in command line tool for agent labeling

--no-hostname-label

flow.register(project_name="tester", labels=["foo"])

When registering a flow that uses prefect.config references I’ve noticed that if the prefect.config is not present the registration throws a Box Error execution.

File "/usr/local/lib/python3.9/site-packages/prefect/utilities/storage.py", line 86, in extract_flow_from_file
    exec(contents, exec_vals)
  File "<string>", line 10, in <module>
  File "/usr/local/lib/python3.9/site-packages/box/box.py", line 501, in __getattr__
    raise BoxKeyError(str(err)) from _exception_cause(err)
box.exceptions.BoxKeyError: "'Config' object has no attribute 'direct'"

I’ve further noticed that if references to config are embedded in tasks that it works. Only when referencing config from a Flow context does this error occur. Is there a workaround for this? It seems strange that I have to install all my environment variables to register a flow when its the Agent that is pushing the envvars into the run-time configuration.

Hello everyone! Are there any Prefect courses currently available? I would be glad if you share the links, thanks!

Hello, I am pretty new with prefect so if this has been answered before I apologize, I couldn't find a solution for my issue anywhere. I have a task that returns a nested list of IDs, something like this:

groups = [
  [1, 3, 5],
  [2, 4, 6],
]

I need to process each ID individually, the problem is that each group of IDs can be parallelized but each group in the list depends on the previous group being fully processed first, like:

for group in groups:
  process_id.map(group) # `process_id` has side effects that affect the result of the next group of IDs

is there a way to process the groups in such a way that the each group is parallelized though

map

but the groups themselves are processed sequentially?

Would appreciate anyone’s insight as to whether this is doable/efficient.

Hi, I need some help or guidance. So right now I have a flow wherein there is a task called "process_document_page" that works on a list of pages in 1 document and processes them using the map. However, now I want to scale and process a list of documents using the same function "process_document_page", which will require a map of map, which I understand is not supported by prefect yet. So is there a solution to create a flow where I can handle this list of documents and call the "process_document_page" on each document? I have something like below, is there a workaround for it?

@task(name="process_document")

def process_document(document): processed_document = process_document_pages.map(document) return processed_document with Flow(name="Local Test") as flow: documents = process_document.map(documents)

We are looking into how to test our flows and even the whole system (think staging vs. production, similar to here: https://prefect-community.slack.com/archives/CL09KU1K7/p1617753993118300) Is there someone who uses physical separation / different machines or accounts for prefect to have isolation between a staging/test environment and production?

Hi, I am having some trouble with running prefect tasks correctly where there is more than one mapped variable ( in addition to unmapped variables). In my case, prefect task is not executing over all combinations of the two mapped different variables, but is iterating over random selection of the same. Is this by design /known behaviour/bug ?

Hello, I'm having some trouble with prefect agent when starting task. I don't really know why this error suddenly occurred, my flow used to work fine and I made some changes on my flow but not on the RunConf...

[2021-04-26 09:20:28,684] INFO - fargate-launch | Found 1 flow run(s) to submit for execution.
[2021-04-26 09:20:31,589] INFO - fargate-launch | Deploying flow run '6021f038-6c79-4e78-862a-e29143f7d205'
[2021-04-26 09:20:31,590] ERROR - fargate-launch | Logging platform error for flow run 6021f038-6c79-4e78-862a-e29143f7d205
[2021-04-26 09:20:34,079] ERROR - fargate-launch | Error while deploying flow
Traceback (most recent call last):
  File "/home/ubuntu/.local/lib/python3.8/site-packages/prefect/agent/agent.py", line 414, in deploy_and_update_flow_run
    deployment_info = self.deploy_flow(flow_run)
  File "/home/ubuntu/.local/lib/python3.8/site-packages/prefect/agent/ecs/agent.py", line 296, in deploy_flow
    taskdef = self.generate_task_definition(flow_run, run_config)
  File "/home/ubuntu/.local/lib/python3.8/site-packages/prefect/agent/ecs/agent.py", line 370, in generate_task_definition
    taskdef["family"] = f"prefect-{slug}"
TypeError: 'str' object does not support item assignment

Do you have any idea ? I didn't try to restart the agent, because I want to find the reason of this failure without resolving it just with a restart. Thanks !

Hey guys, im trying to set up caching for a project i am running locally, with a local prefect server instance. For individual tasks cache works fine. but for mapped tasks the following happens: The first task runs, caches its output. The other mapped tasks just assume the first tasks output is a valid cache, and use that. Here is the result object:

jsonresult = LocalResult(
    location="{flow_id}_{task_full_name}_{flow_run_name}_ignore.json",
    serializer=JSONSerializer(),
)

and the task in question: Hey guys, im trying to set up caching for a project i am running locally, with a local prefect server instance. For individual tasks cache works fine. but for mapped tasks the following happens: The first task runs, caches its output. The other mapped tasks just assume the first tasks output is a valid cache, and use that. Here is the result object:

jsonresult = LocalResult(
    location="{flow_id}_{task_full_name}_{flow_run_name}_ignore.json",
    serializer=JSONSerializer(),
)

and the task in question:

colmatch = MatchColumns(
    result=jsonresult, cache_for=timedelta(days=14)
).map(
    table_metadata=some_list
)

Any ideas why this would happen, and how I can prevent it?

Hi, I try to use the web UI locally but I always get redirected to

<http://universal.prefect.io|universal.prefect.io>

. Does anyone know what's the problem ? Thanks. ps: I've already cleared my cookies (and tried with Chrome incognito mode). ps: I'm using the Docker image

"prefecthq/ui:2021-04-23"

Is there any best practice suggestion/examples on using docker agent + docker run? Right now, after my flow is tested locally, what I do is: • Add one extra line at the end of flow to enable docker-run plus image information, e.g

foo

image • Run a command to register flow to cloud • Build and ship

foo

image based on

prefect/prefecthq

image - in the docker file, I have to copy the code to somewhere like

/app/

and set it as workdir It works, but I am curious about whether somewhere can be improved. Thank you!