prefect-community #prefect-community

argument. I have two different dataframes that I would like to insert into Postgres tables. This is what I have now but unsure of how would I pass the data argument here

```@task

def make_query(schema_name, table_name, columns):

return f'''

CREATE TABLE IF NOT EXISTS {schema_name}.{table_name} ({columns});

INSERT INTO {schema_name}.{table_name} ({columns}) VALUES ();'

'''

with Flow("postgres_test") as flow:

execute_task = PostgresExecute(db_name=database, user=user, host=host, port=port, commit=True)

table_name = Parameter("table_name", default=[df1.name, df2.name])

columns = Parameter("columns", default=[df1_columns, df2_columns])

query = make_query.map(unmapped(schema_name), table_name, columns)

execute_task.map(password=unmapped(postgres_pwd), query=query)

if name == "__main__":

flow.run()```

Leo Kacenjar

02/25/2022, 3:48 PM

Hi there, we are trying to use GitLab storage and the DockerRun flow run. Our Docker image does not contain our gitlab codebase and we'd like to keep it that way. We've almost got everything working, but I'm running into PYTHONPATH issues. I see some similar threads, but none addressing the code outside the image use case. Does anyone have a GitLab Storage/DockerRun combo example? What is the path where the Gitlab repo get exposed to the Docker image? Thanks!

Luke Segars

02/25/2022, 6:32 PM

hi all, i'm working on a flow with a series of mapped tasks -- mapped task 1 feeds results into mapped task 2. i'm observing a strange behavior where: • all child tasks of mapped task 1 complete successfully • all child tasks of mapped task 2 are triggered and complete successfully • BUT for some reason, prefect appears to re-trigger the (already completed) child tasks of mapped task 1 once all have completed. these end up failing (no heartbeat detected, presumably because the dask workers are released b/c the task completed) technically the pipeline completes everything, but all of the children of mapped task 1 are marked as failures once bullet 3 hits ^^^ i'm running this in a gke autopilot cluster...working on testing in another environment now any ideas on why prefect would even try to re-run a successful child task?

Austen Bouza

02/25/2022, 7:42 PM

Hello, I’m seeing an unusual error when submitting flows to ECS. My flows have been running without issue for a long time, but suddenly I’m getting a

prefect.exceptions.AuthorizationError: Malformed response received from Cloud - please ensure that you are authenticated.

message. The auth token I am using has not changed and has not expired. The response is failing on a

requests.exceptions.JSONDecodeError: [Errno Expecting value]

error with the response body

<html><head><meta charset="utf-8"><script>C9xx.T9xx=T9xx;C9xx.g8=window;;n0II(C9xx.g8);T9TT(C9xx.g8);C9xx.D9S=(function(){var Z9S=2;for(;Z9S !== 1;){switch(Z9S)

…and a ton more obfuscated javascript. Is this a typical response body from Prefect Cloud when not authenticated?

Max Lei

02/25/2022, 8:54 PM

If we want to use

DaskExecutor

with

dask_cloudprovider.aws.FargateCluster

, how do we assign a task-definition to the cluster?

iñigo

02/25/2022, 8:57 PM

Hello community I've developed some flows and some of them use the same (made by me) task. I'd like to know if there are some best practices while using a same task over some flows and not repeating in each flow. Thanks in advance

Daniel Komisar

02/25/2022, 9:14 PM

Hello all, when using the dask executor is there any way to control what prefect writes to the console log? We are getting a line in there per prefect log message sent. We send a lot of log messages and it is causing performance problems. Is there any way to turn that off so that the worker doesn’t have to spend the overhead of interleaving lines?

Josh

02/25/2022, 9:31 PM

Is it possible to map over kwargs as asked in this Q&A? https://github.com/PrefectHQ/prefect/discussions/5014 Context: I’m trying to create a cross-product of multiple lists to map over. I want to operate on the downstream values by taking them as kwargs. Inspiration from this question answered by @Kevin Kho https://prefect-community.slack.com/archives/CL09KU1K7/p1645196983284599

Copy code

from prefect import Flow, task
from pprint import pformat


@task
def numbers():
    return [1, 2]


@task
def letters():
    return ["a", "b", "c"]


@task
def cross_product(numbers, letters):
    res = []
    for letter in letters:
        for num in numbers:
            res.append({"num": num, "letter": letter})
    return res


@task
def pretty_print(**kwargs):
    print(pformat(kwargs))


with Flow("...") as flow:
    n = numbers()
    l = letters()
    map_items = cross_product(n, l)
    pretty_print.map(map_items)

flow.run()

I get the following error:

Copy code

Traceback (most recent call last):
  File "flow.py", line 33, in <module>
    pretty_print.map(map_items)
  File "prefect/core/task.py", line 760, in map
    *args, mapped=True, upstream_tasks=upstream_tasks, flow=flow, **kwargs
  File "prefect/core/task.py", line 674, in bind
    callargs = dict(signature.bind(*args, **kwargs).arguments)  # type: Dict
  File "inspect.py", line 3015, in bind
    return args[0]._bind(args[1:], kwargs)
  File "inspect.py", line 2942, in _bind
    'too many positional arguments') from None
TypeError: too many positional arguments

kiran

02/25/2022, 9:33 PM

Does anyone know how to mention

@channel

@here

in the

SlackTask

message? Right now I can tag myself using my user ID but it doesn’t work when I try it with my channel ID or

@channel

<@channel>

Copy code

... 
       elif new_state.is_failed():
            data_engineer_user_id = Secret("DATA_ENGINEER_USER_ID").get()
            msg = f"<@{data_engineer_user_id}> — {obj} ({flow_run_name}): {new_state}"
            SlackTask(message=msg, webhook_secret="SLACK_WEBHOOK_URL").run()

Dexter Antonio

02/25/2022, 9:52 PM

I’m having trouble using python’s

client.graphql

. This is the graphql query that I want to use. It works on the prefect cloud’s interactive api;

Copy code

query {
  flow(where: {name: {_eq: "autoscorer"}, version:{_eq:10}}) {
    id
    name
    version
    tasks{
      name
      id
    }
  }}

This is my attempt to do it with the

client.graphql

Copy code

client.graphql(
{
    'query': {
        'flow(where: {name: {_eq: "autoscorer"}, version:{_eq:10}})': {
            #'id',
            #'name',
            #'version',
            'tasks': {'task_runs': {'id', 'name'}},
        },
    }

})

This works, but if I uncomment the commented fields (e.g.

id

,) I get a syntax error. What is the proper syntax to indicate a root field? Is there any documentation how to use this API in more detail? Is it based off of a common Python GraphQL library?

Aqib Fayyaz

02/26/2022, 7:47 AM

Do we have some working example for docker sidecar on kubernetes on github?

iñigo

02/26/2022, 9:50 AM

Hello, How is possible that A flow.run() works and a run from the server doesn't. I get the next message.

Copy code

Unexpected error: PicklingError('Could not pickle object as excessively deep recursion required.')
Traceback (most recent call last):
  File "/home/inigo/project/scripts/venv/lib/python3.9/site-packages/cloudpickle/cloudpickle_fast.py", line 563, in dump
    return Pickler.dump(self, obj)
RecursionError: maximum recursion depth exceeded while pickling an object

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "/home/inigo/project/scripts/venv/lib/python3.9/site-packages/prefect/engine/runner.py", line 48, in inner
    new_state = method(self, state, *args, **kwargs)
  File "/home/inigo/project/scripts/venv/lib/python3.9/site-packages/prefect/engine/task_runner.py", line 909, in get_task_run_state
    result = self.result.write(value, **formatting_kwargs)
  File "/home/inigo/project/scripts/venv/lib/python3.9/site-packages/prefect/engine/results/local_result.py", line 115, in write
    value = self.serializer.serialize(new.value)
  File "/home/inigo/project/scripts/venv/lib/python3.9/site-packages/prefect/engine/serializers.py", line 73, in serialize
    return cloudpickle.dumps(value)
  File "/home/inigo/project/scripts/venv/lib/python3.9/site-packages/cloudpickle/cloudpickle_fast.py", line 73, in dumps
    cp.dump(obj)
  File "/home/inigo/project/scripts/venv/lib/python3.9/site-packages/cloudpickle/cloudpickle_fast.py", line 570, in dump
    raise pickle.PicklingError(msg) from e
_pickle.PicklingError: Could not pickle object as excessively deep recursion required.

After some testing I also get a problem while publishing my script with flow.register:

Copy code

Traceback (most recent call last):
  File "/home/inigo/project/scripts/venv/lib/python3.9/site-packages/cloudpickle/cloudpickle_fast.py", line 563, in dump
    return Pickler.dump(self, obj)
  File "/home/inigo/project/scripts/venv/lib/python3.9/site-packages/cloudpickle/cloudpickle_fast.py", line 639, in reducer_override
    if sys.version_info[:2] < (3, 7) and _is_parametrized_type_hint(obj):  # noqa  # pragma: no branch
RecursionError: maximum recursion depth exceeded in comparison

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "/home/inigo/project/scripts/es_am_08_caetano_to_s4e/caetano_to_s4e_flow.py", line 65, in <module>
    flow.register(project_name='S4E')
  File "/home/inigo/project/scripts/venv/lib/python3.9/site-packages/prefect/core/flow.py", line 1727, in register
    registered_flow = client.register(
  File "/home/inigo/project/scripts/venv/lib/python3.9/site-packages/prefect/client/client.py", line 1127, in register
    serialized_flow = flow.serialize(build=build)  # type: Any
  File "/home/inigo/project/scripts/venv/lib/python3.9/site-packages/prefect/core/flow.py", line 1489, in serialize
    self.storage.add_flow(self)
  File "/home/inigo/project/scripts/venv/lib/python3.9/site-packages/prefect/storage/local.py", line 143, in add_flow
    f.write(flow_to_bytes_pickle(flow))
  File "/home/inigo/project/scripts/venv/lib/python3.9/site-packages/prefect/utilities/storage.py", line 177, in flow_to_bytes_pickle
    cloudpickle.dumps(flow, protocol=4), newline=False
  File "/home/inigo/project/scripts/venv/lib/python3.9/site-packages/cloudpickle/cloudpickle_fast.py", line 73, in dumps
    cp.dump(obj)
  File "/home/inigo/project/scripts/venv/lib/python3.9/site-packages/cloudpickle/cloudpickle_fast.py", line 570, in dump
    raise pickle.PicklingError(msg) from e
_pickle.PicklingError: Could not pickle object as excessively deep recursion required.

iñigo

02/26/2022, 9:50 AM

Thanks in advance

Hui Huang

02/26/2022, 6:53 PM

Hi, Prefect experts. 😄 We are bootstrapping our analytics stack. I was evaluating different modern solutions. We have production data in SQL database. We want to work on some analytics cleanup & transformation and ship to PowerBI. Based on the research I have done, it looks to me the viable solution for us will be dbt + warehouse(snowflake) + visualization(power bi). If that is the case, it sounds like we do not need prefect to orchestrate python based tasks. Do you have case study that is applying prefect to analytics pipeline achitecture? Hope someone can give me some direction.

Slackbot

02/26/2022, 9:25 PM

This message was deleted.

David Michael Carter

02/26/2022, 10:59 PM

https://docs.prefect.io/core/idioms/task-run-names.html I am having trouble refactoring the example in the documentation above for my needs. I am trying to name tasks based on the mapped children in Prefect Cloud, but I am not mapping a list. I am mapping a dictionary, where the desired

task_run_name

would be a value based on a specific key for that mapped dict index. Anyone have experience naming tasks based on a mapped dictionary?

Ayah Safeen

02/27/2022, 6:41 AM

Hello All.. I was checking the new open-source engine "Orion", but when I run the code I get the following error on the import statement

from prefect import flow

cannot import name 'ThreadedChildWatcher' from 'asyncio'

The code :

Copy code

from prefect import flow
@flow
def common_flow(config: dict):
    print("I am a subgraph that shows up in lots of places!")
    intermediate_result = 42
    return intermediate_result

@flow
def main_flow():

    data = common_flow(config={})

# run the flow
flow_state = main_flow()

discourse 1

Hedgar

02/27/2022, 11:03 AM

I'm glad I didn't give up on myself, I think I have started wrapping my head around this prefect thing anyway thanks to https://medium.com/the-prefect-blog/getting-started-with-prefect-flows-98e8881ffaf9 • I have signed up to prefect cloud • Changed backend to cloud • Registered my project and flow • Done

prefect agent local start

• Was also able to click the

quick run

and I got a beautiful green bar of success! However I have a challenge: My AWS ec2 with the help of lambda function get started at certain time of the day and shuts down at certain time of the day. How do I ensure that my flow runs when AWS ec2 starts for the day? Do I need to create a bash script that start the local agent via a crontab on the remote instance?

Liran Cohen

02/27/2022, 1:49 PM

Hi all, I'm trying to learn Prefect, and I get an error after installation via pip. ModuleNotFoundError: No module named 'prefect'

Liran Cohen

02/27/2022, 1:53 PM

Hi all, I'm trying to learn Prefect, and I get an error after installation via pip. when I run import prefect I get:

ModuleNotFoundError: No module named 'prefect'

the package seems to be installed cause when I do pip show prefect I get:

Copy code

Name: prefect
Version: 1.0.0
Summary: The Prefect Core automation and scheduling engine.

I checked few times and the environment where I try to import is the same where I run pip show any suggestions on why that might happen? thanks

Adi Gandra

02/27/2022, 4:55 PM

Hey, i was trying to figure out is it possible for me to setup a task that if it fails - I can have it retry 3 times (i figured this part out). But, if it still fails I want to kick off another task so the rest of the flow can continue. So basically i’m trying to swap my task if it fails to another task as a backup, so my entire flow can continue on

Adi Gandra

02/27/2022, 9:07 PM

I have my flow setup on Kubernetes (EKS), and everything has been working great. However, sometimes I randomly get “No heartbeat detected from the remote task; marking the run as failed.” and it kills my final task. Some days it works fine, other days the task doesn’t finish and is marked as failed with this error. Any idea’s on this? Or ways to debug what is going on?

Nitin Bansal

02/28/2022, 1:58 AM

Collecting prefect Could not find a version that satisfies the requirement prefect (from versions: ) No matching distribution found for prefect

Nitin Bansal

02/28/2022, 1:59 AM

Getting this error while executing : pip install prefect (on Ubuntu)

Nitin Bansal

02/28/2022, 1:59 AM

Any resolution?

Ovo Ojameruaye

02/28/2022, 6:40 AM

Hi, my state of my flow is set as successful even though task fail. I am not sure why this happens. I have a parent flow which create and runs multiple child flows. A task in a child flow fails but the child flow is marked as successful so other downstream child flows run when they shouldn't. Each child flow (1-5) has a create_flow_run and wait_for_flow_run(). Child flow 2 and 5 wait_for_flow_run() have been set has reference task so the parent flow is successful if they both complete. A task fails in child flow 3 but the flow is marked as successful

Tomer Cagan

02/28/2022, 7:38 AM

Orion question (applies to 1.0 as well I guess): In the docs here, it is stated that tasks can only be called from a flow and not other tasks: "All tasks must be called from within a flow. Tasks may not be called from other tasks." So this means I can only do dynamic task spawning from within the flow function (Flow block in 1.0)? Also, I was working yesterday through the Orion's getting started guide and had an error and found a few typos - where should I report this?

Stephen Lloyd

02/28/2022, 9:06 AM

Hi. I have a flow that I have successfully registered. When I do a quick run, nothing happens. What might I be missing? I followed this tutorial with my own logic. My run_config..

Copy code

RUN_CONFIG = ECSRun(labels=['s3-flow-storage'],
                    task_role_arn=TASK_ARN,
                    execution_role_arn=EXECUTION_ARN,
                    image='<http://1111111.dkr.ecr.us-east-1.amazonaws.com/prefectdemo:latest|1111111.dkr.ecr.us-east-1.amazonaws.com/prefectdemo:latest>',
                    memory=512, cpu=256)

E Li

02/28/2022, 3:34 PM

Hello, does prefect save log files to certain local directory?