prefect-community #prefect-community

Hi all. Some of my flows started failing overnight and after investigation I have found that the EC2 instance that the flows run on has run out of disk space. After a bit of digging I realised that the prefect flow results are filling up the C: drive. What is the best practice for persisting these results? I don't think we really even need them, except maybe in the event of a failure where we might rerun failed jobs.

Carlos Cueto

04/13/2022, 2:09 AM

Hi. I'm new to Prefect and having some trouble understanding the concept of tasks and task runs. I'm trying to return data from a task, but attempting to print the results of the task only prints out

<Task: fetch_data>

. This is what I'm trying to do:

@task

def fetch_data():

return {"data": "random data"}

with Flow('Get-Data') as flow:

flow.run_config = LocalRun()

data = fetch_data()

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

flow.run()

Bihag Kashikar

04/13/2022, 5:40 AM

Hi Guys, I was reading the documentation for running prefect agent on gcp vertex ai and noticed the links on the page here are broken -> screenshot of links

Leanna Morinishi

04/13/2022, 6:27 AM

Hi! Our team wants to use Parameters with default values to record DRIs for Flows (regardless of who created it or updated it last), so that we can query for it. It seems like I have to use a Parameter in a task for it to show up in Prefect UI? Is there a nicer/better way for me to have a Parameter show up in the UI or be graphQL queryable than to write a dummy task that just prints the Parameter?

Jacob Blanco

04/13/2022, 7:01 AM

How can I revoke user API Keys?

Trung Đô Trần

04/13/2022, 8:24 AM

👋 Hello, team!

👋 3

Olivér Atanaszov

04/13/2022, 9:19 AM

Hi, I have a scheduled flow run that had a failed run during the night (probably due to transient network issues). Although Lazarus is enabled for this flow, I did not see any logs reflecting that, see https://docs.prefect.io/orchestration/concepts/services.html#lazarus. Do I miss something?

Stephen Lloyd

04/13/2022, 11:04 AM

I have a flow that when I run locally, works fine. Here’s my flow

Copy code

with Flow(PREFECT_FLOW_NAME, storage=STORAGE, run_config=RUN_CONFIG) as flow:
    creds = get_credentials()
    conn = get_connection(creds)
    tables = get_tables()
    save_data = load_to_s3.map(tables, conn=unmapped(conn))
    conn.set_upstream(creds)
    save_data.set_upstream(tables)
    save_data.set_upstream(conn)

It’s failing on

get_connection

and the relevant code is:

Copy code

@task
def get_credentials():
    return PrefectSecret(PREFECT_FLOW_NAME).run()

@task
def get_connection(creds):
    return connectors.get_redshift_connector(creds)

# from another file...
import redshift_connector


def get_redshift_connector(creds: dict) -> object:
    conn = redshift_connector.connect(
        host=creds['host'],
        database=creds['database'],
        user=creds['user'],
        password=creds['password']
    )
    return conn

When I move to running in ECS, it fails with the following trace…

Copy code

Unexpected error: TypeError("cannot pickle 'Struct' object")
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/task_runner.py", line 930, in get_task_run_state
    result = self.result.write(value, **formatting_kwargs)
  File "/usr/local/lib/python3.8/site-packages/prefect/engine/results/s3_result.py", line 71, in write
    binary_data = new.serializer.serialize(new.value)
  File "/usr/local/lib/python3.8/site-packages/prefect/engine/serializers.py", line 73, in serialize
    return cloudpickle.dumps(value)
  File "/usr/local/lib/python3.8/site-packages/cloudpickle/cloudpickle_fast.py", line 73, in dumps
    cp.dump(obj)
  File "/usr/local/lib/python3.8/site-packages/cloudpickle/cloudpickle_fast.py", line 602, in dump
    return Pickler.dump(self, obj)
TypeError: cannot pickle 'Struct' object

Malthe Karbo

04/13/2022, 12:21 PM

Hi (prefect) - running flows without cloud beta (orion) works fine, but if I login - all works as expected until end of flow, I then receive an error consistenly:

14:17:55.129 | ERROR   | Flow run 'arrogant-yak' - Crash detected! Execution was interrupted by an unexpected exception.

followed by

prefect.exceptions.Abort: This run has already terminated.

. This happens regardless of task success/failure. Again, running without cloud this works fine

Stephen Herron

04/13/2022, 12:40 PM

Hi - we’re experiencing a read_timeout issue with the

invokeLambda

task. I think we need to pass it some extra config through boto_kwargs. Has anyone came across needing to do this before?

Tom Klein

04/13/2022, 1:39 PM

Hello, what’s the best idiom for marking a task of executing some k8s job via

RunNamespacedJob

as a failure (within the job itself) ? or is it impossible because the command is async?

Matthew Seligson

04/13/2022, 2:07 PM

What’s the best way (in Prefect 1.0) to schedule a flow to run at 8am US/Eastern on the second weekday of the month? It seems that CronClocks will handle DST well but can’t handle weekday of the month. And RRule clocks can handle this well but it’s not obvious to me how good their DST handling is. Would appreciate any advice on this!

Tom Klein

04/13/2022, 3:31 PM

Hello again 😄 regarding the

RunNamespacedJob

example (from https://github.com/anna-geller/packaging-prefect-flows/blob/master/flows_task_library/s3_kubernetes_run_RunNamespacedJob_and_get_logs.py ) --- we implemented it and got it to work, but it seems that it’s now failing on :

Copy code

VALIDATIONFAIL signal raised: VALIDATIONFAIL('More than one dummy pod')

because there seems to be many pod “resiudes” of previous runs:

Copy code

['prefect-agent-7745fb9694-6fwk4', 'prefect-job-47d072a8-4pbsf', 'seg-pred-test-cm54l', 'seg-pred-test-doron', 'seg-pred-test-l2j5l', 'seg-pred-test-zvwld']

so wouldn’t k8s keep the pods around given that we gave a “delete_job_after_completion” = False ? and even if the job is deleted successfully, wouldn’t it keep the pods around? or are the pods supposed to be deleted automatically if the job is deleted…?

David Yang

04/13/2022, 3:49 PM

I have a master flow that executes other flows. It runs well so far using git storage. but it failed with this error message: "No heartbeat detected from the remote task; marking the run as failed" tired to first sub flow called in the master flow. Only one line code changed in the master flow. Any suggestions?

Chris Reuter

04/13/2022, 4:00 PM

@Laura Lorenz's livestream starts in 30 minutes! She'll cover an end-to-end prefect2 Prefect 2.0 and cloud2 Prefect Cloud 2.0 experience. You can sign up to watch here.

🦜 2

Kevin Mullins

04/13/2022, 5:02 PM

I’m starting to use

AzureResult

to store task results for my flows. I’ve already got separate storage accounts per environment (dev, qa, prod). I’m curious if it would be recommended to use separate containers for different Prefect projects and/or flows or if it is ok to just store all the results in the same blob container for the environment. My hesitation for further separation is it appears that the AzureResult requires the container to already exist, so I would need to orchestrate something creating containers for each Prefect project. Not a big deal but just trying to get a feel for good practices. Any thoughts appreciated.

Matt Alhonte

04/13/2022, 6:39 PM

How do I delete a bunch of Submitted runs at once? Deleting the Flow didn't seem to do it.

Chris Reuter

04/13/2022, 6:45 PM

Starting in 15 minutes: our 2nd in a 🔥 day of livestreams. @Ben Welsh is the guest on PrefectLive, showing how he uses Prefect and GCP together.

🎉 1

📺 2

🔥 2

Zach Munro

04/13/2022, 7:03 PM

What are the alternative ways to schedule prefect flows besides running them within kubernetes?

Zach Munro

04/13/2022, 7:03 PM

e.g. Is there a Fargate executor?

Zach Munro

04/13/2022, 7:13 PM

One more question: is it possible to have different tasks in a pipeline run in separate containers? For instance, if one task needed to be run on a windows machine, and the next task in the flow needed to be run on a linux machine.

Rajan Subramanian

04/13/2022, 10:15 PM

Hello, i have a fargate eks question. I was in the verge of deploying my prefect file into fargate. I'm using redis as a temporary store. I created a redis elastic cache on AWS and im having trouble connecting to it via kubernetees. Mainly getting a

Copy code

temporary failure in name resolution

I'm not even sure whom to ask this, but since i was using prefect i figured someone here knows about this. I raised it here, https://stackoverflow.com/questions/71864208/unable-to-connect-to-redis-elasticcache-from-fargate, curious if someone had any suggestions? My fargate profile has same 4 subnets that my cluster in elastic cache has. they also have the same security group.

Apoorva Desai

04/14/2022, 12:20 AM

Hello! I have a custom docker container for my prefect flows and I have copied over a python file with some custom functions onto this docker container. However, my prefect flow is unable to import these functions. I am getting a module not found error. How do you access functions from python files in custom docker containers in prefect flow?

✅ 1

Alexander Butler

04/14/2022, 1:57 AM

I am working on standing up Prefect 2.0 is a production environment. For internal data pipeline and reverse etl uses so no fire hazards on my end to use 2.0 early here. Is there a general preference on YAML vs Code for the deployment specification. I noticed you can configure a flow deployment with YAML but I cant find any information on the schema of that document. For example:

Copy code

- name: elt-salesforce
  flow_location: ./salesforce_flows.py
  flow_name: elt-salesforce
  tags:
    - salesforce
    - core
  parameters:
    destination: "gcp"
  schedule:
    interval: 3600

Assuming interval is seconds? Can I specify another grain? Can schedule take a dict? If it takes cron, does that take a dict? Honestly schedule is the primary question point. Everything else is straightforward enough.

discourse 1

Salim Doost

04/14/2022, 2:53 AM

We’re having issues with deployments on one of our Prefect workspaces all of a sudden (without changing anything on our setup to the best of our knowledge). Our flows are stored as Docker images in AWS ECR. Running a newly created flow leads to the following error:

Copy code

404 Client Error for <http+docker://localhost/v1.41/containers/create?name=quantum-squid>: Not Found ("No such image: <account-id>.<http://dkr.ecr.ap-northeast-1.amazonaws.com/datascience-prefect:<image-tag-name>%22|dkr.ecr.ap-northeast-1.amazonaws.com/datascience-prefect:<image-tag-name>">)

However, we’re able to confirm that the image with this tag exists on EMR. Updating an existing flow by overriding an existing image-tag leads to the following error:

Copy code

KeyError: 'Task slug <task-name> is not found in the current Flow. This is usually caused by a mismatch between the flow version stored in the Prefect backend and the flow that was loaded from storage.
- Did you change the flow without re-registering it?
- Did you register the flow without updating it in your storage location (if applicable)?'

Again, we’re able to confirm in AWS ECR that the image got pushed and updated successfully. Our deployment job didn’t throw any error messages as well. Any idea what we can do to resolve this issue?

Carlos Cueto

04/14/2022, 4:27 AM

Hi. I'm having an issue using a

LocalRun

flow's

working_dir

parameter. Whenever I specify the following:

flow.run_config = LocalRun(_working_dir_='C:/scripts/GetADUsers', _labels_=["SVRBIPTH01"])

Whenever I register the flow (I'm using Prefect 1.2.0 on MacOS python 3.10) I get the following working_dir on the UI of Prefect Cloud:

/Users/carloscueto/Documents/Python_Scripts/Prefect-Flows/PowerShell/GetADUsers/C:/scripts/GetADUsers

It seems to be adding the path from where I register the script from (on the local machine) to the working_dir string I specified on the run_config. Has anybody encountered this before? Everything works fine when I register the flow from a Windows computer.

✅ 1

Alexander Butler

04/14/2022, 6:13 AM

One more question thats far more importantly than the YAML flow deployment config... If I want to bundle a self contained application as a docker image which required no ui interfacing to stand up Prefect 2.0, so

prefect orion start

prefect deployment create ...

prefect work-queue create -t etl -- etl-queue

HERE IS THE GAP -- the response to the above command is something like

UUID('...')

, which is useless when setting something up from the CLI without sed/awk?

prefect agent start 'no simple headless way to derive id...'

The less appealing part afterwards is that

prefect work-queue ls

renders a table which is pretty in a CLI but useless again to simply get an ID. Has anyone set up Prefect 2.0 to self deploy in an image along with all their code? The ephemeral nature makes this very advantageous with what seems to be a tiny unconsidered gap. I am pretty sure a more reliable consistent way to get work queue is all thats needed basically, but if I am totally missing it just lmk. I am a big fan of the package for the record but now its crunch time production use attempts 🙂

Jacob Blanco

04/14/2022, 6:32 AM

Anyone else seeing timeouts against the Prefect API? The Cloud view is not updating consistently and many of our Cloud-managed flows are not starting up.

✅ 3

Stéphanie Cérulli

04/14/2022, 6:43 AM

Yes we have errors on API too

Stéphanie Cérulli

04/14/2022, 6:44 AM

Copy code

2022-04-14T06:42:11.384990+00:00 - -  urllib3.exceptions.MaxRetryError: HTTPSConnectionPool(host='<http://api.prefect.io|api.prefect.io>', port=443): Max retries exceeded with url: / (Caused by ReadTimeoutError("HTTPSConnectionPool(host='<http://api.prefect.io|api.prefect.io>', port=443): Read timed out. (read timeout=15)"))