Another question concerning meta data storage of task results: We are thinking of writing a

write_task_result_to_table

task that gets the result of other tasks and writes e.g. into a table whether the task was a

success

,

failed

,

missing data

or empty (= task was not yet executed). We have a flow that runs a several tasks for many systems. So the table we would like to have as result would look like:

system id, task_1, task_2, task_3, ...

and then one row for each system. Is this there already a more elegant way to write all the results of a flow into a table or something like this? Or is the above described way the way to go? Cheers 🙂

Hi friends, I’m having a lot of issues with python modules and packaging when it comes to Docker and testing. Does anyone have a sample project that includes: • A custom Dockerfile for prefect’s docker storage that installs your code as a package • Tests • Setup.py (I’m new to Python’s dependency management)

Hi, I was doing Prefect Server PoC and it works all good and it looks interesting to my manager though when I was trying to do deploy on my company's server, I soon realise that they have blocked the docker hub and they only have their own repository. Having spoke with couple of them they suggest I build the docker image on local server. Not sure where to start, if you can provide any pointer that would be great, thanks.

hey there! within Prefect cloud, how can I change my users email address? i rushed it and typo’d

Hi everyone 👋, Can someone assist with this question please? I simply try to backfill data by passing a datetime parameter https://stackoverflow.com/questions/64029629/is-there-a-way-to-backfill-historical-data-once-for-a-new-flow-in-prefect @Chris White gave an answer, but I can’t get it to work at all.

This message was deleted.

This message was deleted.

Heya! Does anyone know if we are able to control access to specific flows on a member level? For example, can we only allow certain users to interact with and view a specific flow (or project) through the cloud UI? I haven't been able to find this info anywhere in the docs 🤔

Hi ( Prefect Noob here) Do we create a docker image registry per flow when using Docker storage?

Hello - Im running a EC2 VM with prefect docker agent to communicate with Prefect cloud - Right now I have kept ALL TCP (💀 ) Open on the SG to test - for production rollout - I would like to keep only the Prefect Cloud Public Ips open the Security Group Rules - My google fu failed - Can someone link to list of Prefect Cloud Public IP addresses please .

Hi there, is there a TTL for inactive agents, or is there any API documentation available to clean up dead agents. Thanks in advance

Can anyone share their strategy for setting up a Staging-like environment with Prefect Cloud? Given that we only have a single tenant whatever solution we come up with will never truly be totally separate from Production flows, so I was curious what people’s approach is here? I realize that you can do local testing but of course that doesn’t mirror every aspect of the environment when deploying stuff.

Hi, probably a dumb question, but just to be sure, tasks returning generators as part of their output, are they always guaranteed to work ? Something like this : ( note, I will never use a Dask runner, this is a special case working on huge files )

class Process(Task):

def fu(self, rows):

for row in rows:

yield row + 100

def run(self, data):

output = {}

output["meta"] = data["meta"]

output["data"] = self.fu(data["rows"])

return output

@task

def pull(data):

for row in data["data"]:

print(row)

with Flow("test") as flow:

input_data = {"rows": [1,2,3,4], "meta": "fixedvalue"}

process = Process()

data = process(input_data)

drain = pull(data)

Hello everyone! I'm trying to find the right way to send a single slack notification for a group of mapped shell tasks. In my case, the list of mapped task is quite big, therefore, the slack messages are numerous and the slack channel has become really spammy. To be more specific, let's suppose that a group of similar shell commands should be mapped:

from prefect.tasks.shell import ShellTask
from prefect import Flow

a = ShellTask()
commands = [
	'ls',
	'ls -l',
	'invalidcommand'
]

with Flow('test') as flow:
	b = a.map(commands)
	# send a single slack nofiication
	# that summarize the states of mapped tasks
	send_slack_notification()

The purpose of this slack notification is to summarize the states of mapped tasks by displaying the percentage of successfully mapped tasks. In the case above, this percentage will be 66.6% (the last command will fail). I tried to approach the problem using triggers and state handlers but I couldn't find a clean way to achieve the goal. Have you been in this situation before? Any hint? Thanks in advance!

Hi 👋 Does anyone know how (if at all possible) to iterate over a

prefect.Paramter

? My question in more detail: https://stackoverflow.com/questions/64155793/is-it-possible-to-loop-over-a-prefect-parameter

hey all! Does anyone know how to specify these two configuration parameters as environment variables? https://github.com/PrefectHQ/prefect/blob/master/src/prefect/config.toml#L44-L45

Dear all, I just now got the following error:

Unexpected error: OSError("Timed out trying to connect to '<tcp://10.100.0.100:44991>' after 10 s: Timed out trying to connect to '<tcp://10.100.0.100:44991>' after 10 s: connect() didn't finish in time")

Is this a known issue? The environment is a daskkubernetes cluster running on AWS EKS...

Does anyone have a working example of this sentence from the docs: “Package your scripts up into a true Python package. You will most likely need to use the 

COPY

 instruction to put your package into the image, and then the 

RUN

 instruction to install it.”

Hi 👋, If I do a

my_function.map(a, b)

where

len(a)

is 2 and

len(b)

is 5, will this result in 10 runs? Currently I think it only runs twice : / 🤔

hi there - I’m getting an error when trying to run one of my flows:

'413 Client Error: Request Entity Too Large for url: <http://prefect-api.aq.tc/graphql>'

Hi, I 'd like to ask some help/advice on setting up dependent flows. A simplified example of my set up is: I have a flow that ingests some data through a couple of tasks, lets call it flow1. I have another task that I'd like to run from a different flow, call it flow2. The catch is that there will be many such data-ingestion (flow1) dependent tasks. How can i make flow2 have its own schedule but only kick off once flow1 is completed ? I find the solution of making all tasks belonging to one big flow (and setting dependencies there) unsatisfactory as this will grow a lot in the medium term. Thank you in advance for any pointers!

Hi again 👋 I a questions regarding the flow order. Below is a snapshot of a flow I have where I use map and non-mapped parameters. My question is (in the context of my diagram): Why does the flow wait for

get_url_data

to finish, until

transform_url_data

starts? Same goes for the next step. The problem I have is that

get_url_data

will grow quite large and I guess since the information is kept, I can run out of RAM and the whole thing can crash. If the flow kept going iteratively, then the information will be disposed in the process and I have no problem. I must be missing something here I guess and probably there is a better way to design. If anyone can point me in the right direction, it’s much appreciated : )

Hey all we are just about to finish our first prefect flow and have enjoyed the dev experience. I am trying to register my flow in gitlabs CI and push its Docker storage to google container registry, but I am getting two errors I assume revolving around Docker in Docker issues. Please see below

With DOCKER_HOST = <tcp://docker:2376>

    raise DockerException(
docker.errors.DockerException: Error while fetching server API version: HTTPConnectionPool(host='docker', port=2376): Max retries exceeded with url: /version (Caused by NewConnectionError('<urllib3.connection.HTTPConnection object at 0x7fb67baf4760>: Failed to establish a new connection: [Errno 111] Connection refused'))


without DOCKER_HOST set 

docker.errors.DockerException: Error while fetching server API version: ('Connection aborted.', FileNotFoundError(2, 'No such file or directory'))

Hi, is there a way to see if a project already exists?

from prefect import Client
client = Client()
client.create_project(project_name="Hello, World")
# client.get_project(project_name="Hello, World")

When registering a flow, it looks a project is required in later versions of prefect, but for automation purposes I'd like to see if the project already exists before trying to create.

Hi there. I am having trouble with the docker agent. Here is the error I am getting when the agent tries to run the flow:

[2020-10-01 19:07:39,119] ERROR - agent | Error while deploying flow: APIError(HTTPError('400 Client Error: Bad Request for url: <http+docker://localhost/v1.40/containers/create'>))

I think it's related to the

tls_config

argument which defaults to

False

. Is it possible to set this in the command line?

What is the best way to pass several environment variables to a docker agent? I have them in a

.env

file but I don't see a way to pass this file in the cli. Alternatively, how do I tell the agent to use my

config.toml

? I have one at

~/.prefect/config.toml

that is used when using

flow.run()

Can I tell the agent running on the same machine to use this config file?

Does anyone know why is this an invalid format in the Prameter section of the UI?

{
"strings": ['string1', 'string2'],
'start_date': '20200920',
'end_date': '20200930',
}

Hi! I'm looking for good design pattern for unit testing tasks. I've seen the examples that compose and test flows but it makes more sense to me to start with a test for each of my tasks instead.

Is it possible to set the

dir

for a

LocalResult

at run time? I am setting it based on a context variable but it seems to be getting the value of the variable at registration time.

@task(
    target="ftp_{start_date}_{end_date}.done",
    result=LocalResult(dir=prefect.context.local_results_dir),
)
def mytask(start_date, end_date):
    pass