prefect-community #prefect-community

Hi there, I'm wondering if there is a recommended way to include sql scripts as part of a flow? Perhaps using a custom base image with the sql files already loaded?

jeff n

01/15/2021, 11:16 PM

Hello all. I have a flow that can take parameters such as db table and pull data from that table generically. How do I register the same flow to run in prefect cloud with different parameters. For instances the flow pulls from

projects

with one set of runs and

accounts

with another set of runs but both of the runs are the same flow code and need to run separately.

BK Lau

01/16/2021, 12:31 AM

Q: if i successfully run a Flow/Task locally and now wanted to run the same flow/task in another remote cluster and I "register" the Flow/task with the Prefect server, its my understanding that only the Flow/Task DAG metadata got serialized to the Prefect Server/Database. My question is how does the full complement of the Flow/Task code and its third-party dependency libraries got deployed or loaded in the remote cluster? Am I missing something here? Who is the deploying the code to the remote cluster??

Riley Hun

01/16/2021, 12:40 AM

Hello all - I feel like I'm super close to getting my flow productionized and integrated with Dask Gateway. Unfortunately, came across a

PermissionError

(see reply for full stacktrace of error). Tried to look through the thread to see if others have encountered a similar error, but no luck. Any insight on this?

Sonny

01/16/2021, 1:44 AM

whats a right way to submit a spark job from prefect task?

Hui Zheng

01/16/2021, 3:35 AM

Hello, The prefect cloud seems unstable tonight, We have seen runs get stuck or not connecting to Prefect graphql, and also the prefect dashboard was loading very slow in the past few hours.

Amanda Wee

01/16/2021, 11:56 AM

If the flow registration

idempotency_key

matches the one from the previous flow registration, the flow version is not bumped. However, does the serialised flow get uploaded to storage anyway? I'm too newbie at the codebase (and graphql in particular) to make sense of the details of the flow registration code.

👀 2

Tadas

01/16/2021, 3:51 PM

How can I give mapped tasks specific task_run_names?

Marwan Sarieddine

01/16/2021, 4:55 PM

Hi folks, question about inspecting prefect function tasks, is there a way to retrieve the arguments that are used to run the task. Please see a simple example in the thread.

jack

01/18/2021, 2:30 AM

Hey Prefect team, I had a question - when we are passing data between flows or getting data through tasks, are we ever storing that data in Prefect's Data Centers or infrastructure even for a small amount of time? I guess this is more of a concern for sensitive PI data (for issues like GDPR etc.) Cheers!

Aiden Price

01/18/2021, 7:16 AM

Hi Prefect people, I get an occasional error where my Kubernetes agent doesn't deploy a flow run, I'm using the Helm deployed Prefect Server v14.3. I can see this error in the agent's logs;

Copy code

[2021-01-17 00:00:11,063] ERROR - Prefect-Kubed | Error while managing existing k8s jobs
Traceback (most recent call last):
  File "/usr/local/.venv/lib/python3.8/site-packages/prefect/agent/kubernetes/agent.py", line 362, in heartbeat
    self.manage_jobs()
  File "/usr/local/.venv/lib/python3.8/site-packages/prefect/agent/kubernetes/agent.py", line 219, in manage_jobs
    event.last_timestamp
TypeError: '<' not supported between instances of 'NoneType' and 'datetime.datetime'

Sven Teresniak

01/18/2021, 9:21 AM

Hi, after upgrading from 0.14.1 to 0.14.3 I got the error

ValueError: Multiple flows cannot be used with the same resource block

. In which direction I have to search for a solution? We heavily rely on a ResourceManager available for multiple flows…

Adam Roderick

01/18/2021, 12:42 PM

Hi! I am looking for a way in prefect cloud to report on successful/failed flow runs over the last week, month etc. What do you recommend? I want to filter by label and date range. Any suggestions? Thanks!

Matic Lubej

01/18/2021, 1:46 PM

Hello! I have a question about creating complex prefect flows. Is there a prefect native way to achieve a nested tree-like process, where multiple tasks are spawned from single tasks and some of them are later merged again? IDK how to best describe it and I'm not sure how detailed my description should be on this channel, but perhaps this sketching helps

Matic Lubej

01/18/2021, 2:05 PM

And one other, I guess independent thing. When I tried to run the [flat-mapping](https://docs.prefect.io/core/concepts/mapping.html#flat-mapping) example and vizualized the graph with the flow state, I get the following

Matic Lubej

01/18/2021, 2:06 PM

I'm guessing that the parts on the right should come out of each specific B blocks?

Kieran

01/18/2021, 2:40 PM

Hey 👋 , I am new to Prefect Cloud and have hit a task definition snag which I can't seem to solve. We have an ECS Agent deployed on EC2 with the correct API tokens which shows up in our Cloud UI. We have a ECS cluster which we declare when launching the agent but leave the launch type as default (Fargate). Our example Flow is below:

Copy code

default_client = docker.from_env()
FLOW_NAME = "hello-flow"
flow_schedule = CronSchedule("0 8 * * *")
flow_storage = Docker(
    base_url=default_client.api.base_url,
    tls_config=docker.TLSConfig(default_client.api.cert),
    registry_url="<http://_________.dkr.ecr.eu-west-2.amazonaws.com/xxxxx/prefect|_________.dkr.ecr.eu-west-2.amazonaws.com/xxxxx/prefect>"
)
flow_run_config = ECSRun(
    cpu="512",
    memory="512",
    run_task_kwargs={"requiresCompatibilities": ["FARGATE"], "compatibilities": ["FARGATE"]}
)

with Flow(
    name=FLOW_NAME,
    schedule=flow_schedule,
    storage=flow_storage,
    run_config=flow_run_config
    ) as flow:
    say_hello()

if is_serializable(flow):
    flow.register(project_name="Test", registry_url=flow_storage)
else:
    raise TypeError("Flow did not serialise.")

We are getting the following error from our task logs:

Copy code

An error occurred (InvalidParameterException) when calling the RunTask operation: Task definition does not support launch_type FARGATE.

In an attempt to resolve this issue I tried adding

run_task_kwargs

as above but with no luck. Does anyone have any pointers? (I can see from the ECS task definition panel that the
Compatibilities
is set to EC2 and
Requires compatibilities
is blank and from this thread that could be the cause...)

Jeff Williams

01/18/2021, 9:51 PM

Can anyone tell me where to find information about Prefect and TLS support?

Sai Srikanth

01/18/2021, 11:04 PM

Hey Prefect team, I am trying to connect to AWS? I am trying to integrate by passing ACCESS_KEY and SECRET_KEY but I couldn't.

Felix Vemmer

01/18/2021, 11:57 PM

Hi everyone, I am getting the following error, when trying to write a pandas dataframe into google Cloud Storage:

Copy code

Unexpected error: TypeError("__init__() got an unexpected keyword argument 'client_options'")
Traceback (most recent call last):
  File "/Users/felixvemmer/.pyenv/versions/3.8.6/envs/automation_beast/lib/python3.8/site-packages/prefect/engine/runner.py", line 48, in inner
    new_state = method(self, state, *args, **kwargs)
  File "/Users/felixvemmer/.pyenv/versions/3.8.6/envs/automation_beast/lib/python3.8/site-packages/prefect/engine/task_runner.py", line 891, in get_task_run_state
    result = self.result.write(value, **formatting_kwargs)
  File "/Users/felixvemmer/.pyenv/versions/3.8.6/envs/automation_beast/lib/python3.8/site-packages/prefect/engine/results/gcs_result.py", line 77, in write
    self.gcs_bucket.blob(new.location).upload_from_string(binary_data)
  File "/Users/felixvemmer/.pyenv/versions/3.8.6/envs/automation_beast/lib/python3.8/site-packages/prefect/engine/results/gcs_result.py", line 41, in gcs_bucket
    client = get_storage_client()
  File "/Users/felixvemmer/.pyenv/versions/3.8.6/envs/automation_beast/lib/python3.8/site-packages/prefect/utilities/gcp.py", line 53, in get_storage_client
    return get_google_client(storage, credentials=credentials, project=project)
  File "/Users/felixvemmer/.pyenv/versions/3.8.6/envs/automation_beast/lib/python3.8/site-packages/prefect/utilities/gcp.py", line 31, in get_google_client
    client = Client(project=project, credentials=credentials)
  File "/Users/felixvemmer/.pyenv/versions/3.8.6/envs/automation_beast/lib/python3.8/site-packages/google/cloud/storage/client.py", line 122, in __init__
    super(Client, self).__init__(
TypeError: __init__() got an unexpected keyword argument 'client_options'

I am running a task thats returning a

pd.DataFrame

which I am trying to store into Google Cloud Storage:

Copy code

pandas_serializer = PandasSerializer(
    file_type='csv'
)

gcs_result = GCSResult(
    bucket='tripliq-data-lake',
    serializer=pandas_serializer,
    location=f'linkedin_top_posts/{datetime.datetime.now().strftime("%Y%m%d-%H%M%S")}_linkedin_post_likes.csv'
)

like_linkedin_feed = LikeLinkedInFeed(
    result=gcs_result
)

I am not understanding the source code too well, but I think it’s referring to this line in

site-packages/google/cloud/storage/client.py

def __init__(

self,

project=_marker,

credentials=None,

_http=None,

client_info=None,

client_options=None,

):

Any help is very much appreciated!

Alex Rud

01/19/2021, 3:42 AM

Hi… Is there a way to map over a dataframe without producing some intermediary objects that bloat memory?

Greg Roche

01/19/2021, 2:45 PM

Hi folks, does anyone have any experience with this error when running a

LocalDaskExecutor

flow, using a LocalAgent running inside a Docker image?

TypeError: start() missing 1 required positional argument: 'self'

Edit: solved, I wasn't initialising the LocalDaskExecutor.

Copy code

flow.executor = LocalDaskExecutor  # wrong
flow.executor = LocalDaskExecutor()  # this works

Joël Luijmes

01/19/2021, 3:29 PM

How can I make a distinction (run-time) to see wether I’m running the flow directly or from the agent / server? Case: when running on production (i.e. from the kubernetes agent) it should perform a task (actually its a resource manager) - which deploys a cloudsql proxy and connect through that proxy - but if I’m developing I want to connect with localhost database. In theory, its fine if I deploy the cloudsql proxy (resource manager) while developing the flow, but I still need to know at runtime if which hostname it should use to connect with the database.

Pedro Machado

01/19/2021, 4:07 PM

Hi everyone. I tried to set up the docker agent with

supervisord

a while ago and was getting a permission error when trying to access the docker engine. I never got it to work. I'd like to run the Docker Agent inside of a container managed by docker compose. I have a couple of questions about this set up: 1. I am setting the

restart

policy to

always

Would this be enough to restart the agent if it failed or in case of restart of the host? 2. What is the best way to give the agent access to the docker engine running on the host? Thank you!

01/19/2021, 5:28 PM

Someone, please share AWS setup for prefect. I did "pip install prefect" but still getting the error

01/19/2021, 5:28 PM

[xxxxxserverxxxxxx]$ python prefect_sample.py Traceback (most recent call last): File "prefect_sample.py", line 5, in <module> import prefect ImportError: No module named prefect

BK Lau

01/19/2021, 5:44 PM

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

Matic Lubej

01/19/2021, 8:18 PM

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:

Copy code

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!

Billy McMonagle

01/19/2021, 9:29 PM

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

Chris Jordan

01/19/2021, 10:13 PM

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

Copy code

@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

Copy code

@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?

👀 1