Q: Can Prefect Agent be set up to run Hashicorp

Nomad

as executor instead of

Dask

?? I want to use Prefect on the Cloud to drive a

Nomad

cluster installed on-premise

Hi again! I'm trying to run a process over a Fargate cluster using

dask_cloudprovider

API. running tutorials and sample code from

dask

this works great, but for prefect I have created a dedicated docker image which I provide to the cluster initializer. The cluster gets created, but as soon as the flow starts, after 10 s I get the following time-out error:

Traceback (most recent call last):
  File "/usr/local/lib/python3.8/site-packages/prefect/engine/runner.py", line 48, in inner
    new_state = method(self, state, *args, **kwargs)
  File "/usr/local/lib/python3.8/site-packages/prefect/engine/flow_runner.py", line 418, in get_flow_run_state
    with self.check_for_cancellation(), executor.start():
  File "/usr/local/lib/python3.8/contextlib.py", line 113, in __enter__
    return next(self.gen)
  File "/usr/local/lib/python3.8/site-packages/prefect/executors/dask.py", line 203, in start
    with Client(self.address, **self.client_kwargs) as client:
  File "/usr/local/lib/python3.8/site-packages/distributed/client.py", line 748, in __init__
    self.start(timeout=timeout)
  File "/usr/local/lib/python3.8/site-packages/distributed/client.py", line 953, in start
    sync(self.loop, self._start, **kwargs)
  File "/usr/local/lib/python3.8/site-packages/distributed/utils.py", line 340, in sync
    raise exc.with_traceback(tb)
  File "/usr/local/lib/python3.8/site-packages/distributed/utils.py", line 324, in f
    result[0] = yield future
  File "/usr/local/lib/python3.8/site-packages/tornado/gen.py", line 762, in run
    value = future.result()
  File "/usr/local/lib/python3.8/site-packages/distributed/client.py", line 1043, in _start
    await self._ensure_connected(timeout=timeout)
  File "/usr/local/lib/python3.8/site-packages/distributed/client.py", line 1100, in _ensure_connected
    comm = await connect(
  File "/usr/local/lib/python3.8/site-packages/distributed/comm/core.py", line 308, in connect
    raise IOError(
OSError: Timed out trying to connect to <tcp://172.31.40.184:8786> after 10 s
[2021-01-19 20:14:19+0000] ERROR - prefect.Execute process | Unexpected error occured in FlowRunner: OSError('Timed out trying to connect to <tcp://172.31.40.184:8786> after 10 s')
Traceback (most recent call last):
  File "s3_process_l2a_2019.py", line 114, in <module>
    assert status.is_successful()
AssertionError

Any ideas what is going on? Is the dask scheduler having issues connecting to the workers? Or might it be something else? Thanks!

I have a flow with docker storage, running with the k8s agent. When I run my flow and look at the logs, it seems to be executing an old version of my flow. Details in thread...

Hey, I've got a quick question about persisting temporary files between tasks. It looks like if I pass a

tmpfile

object between tasks, it'll work as expected, as in

@task(name="save records to a tmp file")
def save_some_data(result=PrefectResult()):
    logger = prefect.context.get("logger")
    the_file = tempfile.NamedTemporaryFile()
    for i in range(5000000):
        the_file.write(b"lorem ipsum\n")
    the_file.seek(0)
    <http://logger.info|logger.info>("wrote the file")
    return the_file

@task(name="reread that data")
def read_some_data(the_file, result=PrefectResult()):
    logger = prefect.context.get("logger")
    output = the_file.read()
    <http://logger.info|logger.info>(f"read the file")
    <http://logger.info|logger.info>(f"length of file is {len(output)}")


with Flow("save_retrieve_file_flow") as flow:
    f = save_some_data()
    g = read_some_data(f)

but passing the name of the file won't work - the file will not be found

@task(name="save records to a tmp file")
def save_some_data(result=PrefectResult()):
    logger = prefect.context.get("logger")
    the_file = tempfile.NamedTemporaryFile()
    for i in range(500000):
        the_file.write(b"lorem ipsum\n")
    the_file.seek(0)
    <http://logger.info|logger.info>(f"wrote the file to {the_file.name}")
    return the_file.name

@task(name="reread that data")
def read_some_data(the_file_name, result=PrefectResult()):
    logger = prefect.context.get("logger")
    <http://logger.info|logger.info>(f"trying to open {the_file_name}")
    
    with open(the_file_name, 'r') as the_file:
        output = the_file.read()
        <http://logger.info|logger.info>(f"read the file")
        <http://logger.info|logger.info>(f"length of file is {len(output)}")


with Flow("save_retrieve_file_flow2", 
        state_handlers=[cloud_only_slack_handler]
       ) as flow:
    f = save_some_data()
    g = read_some_data(f)

what's going on here? is the file system being reset between tasks? does prefect clean up temporary files in the flow if they're not in memory directly? something else?

Hello Prefect, One our flow run got an Connection Error at the start of the flow run, which we never seen before. Do you know what caused that?

HTTPSConnectionPool(host='10.124.0.1', port=443): Max retries exceeded with url: /apis/batch/v1/namespaces/default/jobs (Caused by NewConnectionError('<urllib3.connection.HTTPSConnection object at 0x7fa7a7705668>: Failed to establish a new connection: [Errno 111] Connection refused',))

Hi Prefect Team, I upgraded to Prefect version

0.14.4

and I can no longer run

flow.environment.executor = executor

. I get the following error

AttributeError: 'NoneType' object has no attribute 'executor'

. Does anyone know how to attach an executor to a flow registered to Prefect Server using Docker Storage in the latest version of Prefect. Had no issues with this is Prefect version

0.13.18

. In the meantime, I'll downgrade to

0.13.8

for the time being.

I have setup prefect in aws ec2 and able to register the flow and the flow is running too, but the aws prefect server page is not showing any flow and the page looks like the below. Please help

Hi Prefect people, I'm wondering how I can get the KubernetesRun to use a specific image tag that I want to provide dynamically (or at least whenever I reinstall the server and agent with Helm). The backstory is that I have a custom base image for my Agent, Kubernetes jobs and Dask cluster, and when I launch a KubernetesRun there's a delay before the job starts because I call for the

latest

tag and

Always

pull so I can be sure I'm getting the latest dependencies. I'd like to minimise the delay by changing to a

IfNotPresent

image pull policy, but then I'd need to pin the image tag to a specific version, but then I'd want to make sure that specific tag could be updated as I keep pushing updates. Does anyone have any advice? Thanks.

Hi Prefect Team, just a quick (probably silly) question. Please pardon my ignorance. I upgraded to

Prefect 0.14.4

. My flow, which is using Docker storage and Dask Gateway as the executor, is working fine after the transition. I finished reading the updated documentation and I'm trying to understand the concept of

Run Configuration

. I don't quite understand it, especially since my scheduled flow worked without it, so how do I know when a Run Config is applicable?

Hello there, is there a way to register flow from within another flow, or clone existing flow with changed name. We need to re-schedule flows each night for next day, different customers, the idea is to register flow with name which includes customer name, or clone existing flow with such name. The problem is that registration of flow from within another running flow was not working in previous versions, not sure what is the current status. And cloning with another name and parameters is preferable. Any ideas?

Hi all, we are seeing the error "At least one upstream state has an unmappable result" for a failed task. A subsequent manual run of the same flow finishes without error. We think this might only happen when the upstream task has to retry several times before succeeding, but I've failed to reproduce this. Any idea what might be causing this?

Hi folks, it seems I am facing a bug when trying to pass parameters returned from a task to

StartFlowRun

- I just want to make sure I am doing this correctly - please see the example in the thread.

How to re-schedule flow which already exists? We have scheduler job, and registered flow, is it possible to re-schedule it ?

Hello all, I have seemed to come across a bug, or at the very least something interesting. I have been having issues with getting the graphql service seen here to connect and I have discovered an oddity in that I had a config.toml and a config.toml.bak and the best that I can tell is that Prefect had attempted to parse both files leading to issues that was resolved after I removed the .bak file entirely. Has anyone else encountered this?

I am planning to run large Flow with the caching that seems to be working fine But lately I found out that if the same flow is started at almost similar time then the caching might not work as they both are running at the same time. Want to understand if there is a way to control this by not triggering the second flow when the first run is already in running state (with the same Parameter) using Prefect API or GraphQL query API

Hello all. I have 2 tasks in a flow, execute_a and execute_b. The task execute_a is using raise LOOP to make successive calls to an API and will return a string when it finishes. When I simulate an error (eg. API is down), I notice that execute_a is set to Retry state and execute_b is also started and set to Pending state (I was expecting execute_b to not even start). At this point the API will become available again and eventually both tasks will finish with success. My question here is, how can we tell Prefect to only start execute_b if execute_a is finished?

Hi there, I have a few months with prefect. Can yo help me with this question?: I have the following flow. 1.- 300 csv (with different name and size) -> gzip -> EXASOL(DW) -> Delete (csv, gz) 2.- A flow that proccess 1 csv per run My question is how handle Prefect the flows and server resources if i run al csv at the same time. That's means run 300 flow's at the same time. (Does some kind of smart upload so don't collapse the server?) Enviroment info: # Server 24 Cores - 96GB RAM # Single Node # Total size csv (150GB aprox) # Prefect Core 0.14.0 # Docker Prefect core-0.14.0 # Executor LocalDaskExecutor (You can give some advice with DaskExecutor too) Note: I avoid pass a list of the csv in each task because disk space. Notice that i delete the file at the end of the flow Thanks for your help 🙂

Hi, I am trying to access the registered flow from aws ec2 public dns...something like this http://ec2-xxxxxxxxxxxx.com:8080/default/flow/4dfbfb93-e00a-4ad4-a34d-fdd813f9a123, but the url reverts back to http://ec2-xxxxxxxxxxxx.com:8080 and the dashboard is not showing the flows, but the flow is running and I can see the output being stored in my database...Am I missing something in the setup when the host is not the localhost for the prefect server? @nicholas can you please help on this?

I submitted a few hundred runs using the prefect API that I want to cancel right now, how can I do that easily?

I want to cancel every currently scheduled run for one of my flows basically

One way of doing this is to register a new version of that flow, but that seems hacky

Q: If I have 100 on-prem worker nodes, say, do I have to install a Prefect Agent on each of them to enable them to be controlled by a Prefect Server?

Hi community, Do you have any idea on how to make a task mapping with two parameters? For instance, parameter1 = [a, b, c] and parameter2 [1,2,3] So how could I make tasks with all the pairs possibilities : (a, 1),(a,2),...? Thanks in advance

do any of the official prefect docker images have any of the extras pre-installed in them or is it on you to manually install, for example, the

aws

extras?

Hello there, what is the right way to pass results from task to another as inpurt for run method, when task are dynamically added to edge?

Hi! Quick question. I'm running a prefect flow using the DaskExecutor, which is connected to a cluster. My checkpointing is turned on, but there is no result being save and no logs being printed out (DEBUG mode on). I get the results and the print-out if I use the local executor or a local instance of a DaskExecutor. Any ideas what to do in this case?

Hello all, I am trying to connect Prefect, AWS thru storing secrets as ACCESS_KEY, SECRET_KEY, SECURITY_TOKEN. I am trying to connect as below but I am unable to connect. Any suggestions? Please check the image for further reference:

Hi, Quick Question: How i can use the parameter input as a string outside a task. When i try it , i get a parameter object <Parameter: example>

with Flow("example-flow") as flow:
	example_param = Parameter('example')

	print(example_param)

Thanks for your help!!