Hi there! I was wondering what is the best way to create a prefect task out of a function that is applied to each row of a dataframe. E.g. wrap something like

df = df.apply(my_function, axis=1)

with prefect. Thanks!

Hi there. I have a flow that will run on Kubernetes. I'd like to see log messages in near real time. Currently, the messages show at the end of the run? I wonder if this has to do with the stream not being flushed frequently. How can I flush the stream after calling

<http://logger.info|logger.info>()

? Thanks!

Hey community, In the python api you have the ability to run a flow locally, that has a schedule by adding

flow.run(run_on_schedule=False)

. I can't find this option for the cli when running

prefect run -p my_flow.py

What am I missing?

Hey Prefect, I am getting an error on my flow in prod of

Error uploading to S3: An error occurred (AccessDenied) when calling the PutObject operation: Access Denied

This is from my S3Result object trying to persist the results to the bucket. Not sure why this is happening as my

task-role

has full S3 access and I have an identical set up in staging, yet I dont get these errors. Am I right in assuming that the S3Result just assumes the

task-role

? Why would boto3 not be picking this up?

The only other thing I can think of is if the bucket value parameter is fixed at register time and not at run time of the flow?

and that was the answer.... good to know. Hope that helps someone else out!

Hi, let’s say I’m using Kubernetes agent, at the moment there is a flow running. CI/CD at this moment re-deploy the agent, what will happen with the running flow?

Hi at all, does anybody how to trigger a specific task in a flow with a certain delay of some minutes? thanks :)

Hi everyone! @Aaron Prescott and I are trying to set up our first flows on Prefect cloud using an aws EKS agent. Right now we are getting stuck when trying to specify the storage in aws ECR. When I run

storage = Docker(registry_url="https://<our account id>.<http://dkr.ecr.eu-west-1.amazonaws.com/data/prefect-application/%22|dkr.ecr.eu-west-1.amazonaws.com/data/prefect-application/">)
storage.build()

I get this error

[2021-07-13 14:32:00+0100] INFO - prefect.Docker | Building the flow's Docker storage...
invalid reference format
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/Users/hilaryroberts/.pyenv/versions/3.8.11/lib/python3.8/site-packages/prefect/storage/docker.py", line 303, in build
    self._build_image(push=push)
  File "/Users/hilaryroberts/.pyenv/versions/3.8.11/lib/python3.8/site-packages/prefect/storage/docker.py", line 369, in _build_image
    raise ValueError(
ValueError: Your docker image failed to build!  Your flow might have failed one of its deployment health checks - please ensure that all necessary files and dependencies have been included.

It should be using the default image from the prefecthq dockerhub, so I wonder why it's failing to build. Anyone know what I might be doing wrong?

What's the recommended way to insert data into a postgresql table?

Hello, everyone! I'm beginner in Prefect. I try to build flow which will be running some sql file from GitHub. Using Prefect Secrets I have connect between my Perfect Cloud and my GitHub repo. I try to recieve file this way:

from prefect.storage.github import GitHub
my_string = GitHub(access_token_secret="github", repo="my_repo", path="etl/query.sql")

but I get only

<class 'prefect.storage.github.GitHub'>

I need to get this file like a string. How I can do it?

is there a less hacky way to create dynamic defaults for parameters as opposed to doing something like this?

@task(name="set-date-params", nout=2)
def set_date_params(
        start_date: typing.Optional[str],
        end_date: typing.Optional[str]
) -> typing.Tuple[datetime.date, datetime.date]:
    """
    If start date and end date are explicitly set on the Flow, we simply construct date objects and return them.

    If start_date is not set, we default it to 7 days from the local context date of the flow.

    If end_date is not set, we default it to the local context date of the flow.
    """
    context_run_datetime = context.get("date")
    aware_utc_run_datetime = context_run_datetime.replace(tzinfo=utc)
    aware_et_run_datetime = aware_utc_run_datetime.astimezone(et)

    if not start_date:
        seven_days_ago = aware_et_run_datetime - datetime.timedelta(days=7)
        start_date = seven_days_ago.date().isoformat()

    if not end_date:
        end_date = aware_et_run_datetime.date().isoformat()

    <http://logger.info|logger.info>(f"start_date={start_date}, end_date={end_date}")
    return datetime.date.fromisoformat(start_date), datetime.date.fromisoformat(end_date)


with flow:
    start_date_, end_date_ = Parameter("start-date", default=None), Parameter("end-date", default=None)
    start_date_, end_date_ = set_date_params(start_date_, end_date_)

Hi community 🙂, I’m having a strange bug with my flows on k8s: When a task fails, the flow goes into a failed state but I don’t see the error in the logs only some pod error message and a heartbeat related failure:

Pod prefect-job-9f127532-nnd2f failed.
	Container 'flow' state: terminated
		Exit Code:: 1
		Reason: Error

No heartbeat detected from the remote task; marking the run as failed.

But when I look at the pod logs, I can see the real error. In addition to that, we have no slack notification when the flow fails like that. Any idea how to overcome this ?

Hi everyone, I have a common set of concerns (configuration, logging) I want to setup for each task and my first idea was to override the .run() method of the prefect.tasks.core.function.FunctionTask and use prefect @task decorator with the new FunctionTask, but this method is not being called in the runtime when executing a task? Is it possible to add some cross cutting concern code before running a task itself? Tried a decorator, but seems it is not pickable by cloudpickle. I'll add a code example in the comment

Hi! Whenever the instance running prefect tasks crashes, hearbeats are not detected anymore and the run should fail. Instead of this expected behavior only the task currently running fails, the next ones are still

Pending

and the run shows as

In progress

until I manually cancel it. Did anyone already run into this issue?

No heartbeat detected from the remote task; marking the run as failed.

Hi everyone — is there a way to register a flow with Docker storage from within a docker container? I’m thinking if there’s something clever I can do but not sure

Hello All, I am trying to register my flow and getting an error "ValueError: Project prod not found. Run

prefect create project 'prod'

to create it."

Hey all, has anyone had any experience in using Prefect @task decorator in a python package that can be imported and used in flow scripts? The use case here would be a common database connection function, that inside the function would use the Prefect AWS Task to connect and pull credential information from AWS Secrets Manager

Hi, interested in using the GraphQL API to set off a flow run with parameters. If the following is the query in the post request to start off a flow run, how would I add in any parameters I’d like to specify? Thanks!

query = """mutation {
    create_flow_run(input: { 
    flow_id: "5ac80adc-442a-4e53-bc01-d9d65bffd6aa"}) {
        id
    }
    }"""

Hey all, We have the docker image stored on a private AWS ECR. This seems to stop working after 12 hours of starting the agent, because the authorization token expires and would somehow need to be re-authenticated with aws ecr docker login command in the command line. Otherwise, the docker agent can't pull the latest image on flow run. I've tried passing in the environment variable "PREFECT__CONTEXT__SECRETS__AWS_CREDENTIALS"_ on starting the agent, but that still doesn't seem to solve the issue. Any ideas on how to solve this? NOTE: AWS Authorization Tokens can't be changed from the 12 hour expiration. So If anyone has a way that they solved this for the prefect docker agent, that would be very helpful

Hi everyone I just deployed a flow with Docker storage and ECS Agent, and on my dockerfile I do:

ENTRYPOINT ["make", "deploy"]

and when I ran the flow no logs came to the UI at all (as expected, I think), because the container threw an error with “make” (probably because directory config and etc so no worries about it) so the flow is still in the “submitted” state, even though the actual ECS task has already failed. Is there any way to have better visibility of when that happens? in case a dev forgets to remove the entrypoint from the dockerfile..

Not sure if this is the right spot for this but it would be really helpful if the aws BatchSubmit task printed the error message on this line rather than just catching it

Hello - I was wondering if there is a way to prevent DockerStorage from being built / pushed when the Flow is already registered with a known idempotency_key. Ideally, the storage should only be built/pushed when the Flow has changed.

Does anyone know about this issue?

prefect backend server
Traceback (most recent call last):
  xxxxx
  File "pyev/lib/python3.6/site-packages/prefect/utilities/notifications/__init__.py", line 1, in <module>
    from prefect.utilities.notifications.notifications import callback_factory
  File "pyev/lib/python3.6/site-packages/prefect/utilities/notifications/notifications.py", line 13, in <module>
    from toolz import curry
ModuleNotFoundError: No module named 'toolz'

Hi there, does anyone have an example using greatexpectations tasks? Thanks!

Hi everyone — I’m using Docker storage but I want to register the same flow against 2 Prefect Servers (one is PROD the other is UAT and we have some other as well). All docker images are stored in the same place so we don’t need to build and push the docker image all the time. Is there a way to re-use the docker storage to deploy to many servers? I tried just “not building” because it was already built but I got the error:

Failed to load and execute Flow's environment: ValueError('Flow is not contained in this Storage')

This message was deleted.

Is there any facility to pass a

--build-arg

for a Dockerfile to the prefect register CLI command?

Hello, i was wondering if there's any way we can look at the output of each step using prefect or something similar to the python fn-graph (Python modelling pipeline library) ?

Hi I am using DockerRun for my Flow's environment and I am running into some issues. The Docker image pulls and runs fine, but it uses pytorch and I am starting to run into some shared memory issues for my dataloader. Normally using Docker, people have solved this by passing arguments to increase the shared memory size to the docker run command. Is there any way to change the shared memory of the DockerRun on Prefect?