Another question, if I run a local agent, how can I specify the URL for the prefect API i wish to connect to, if it does not run on the local server? (not using prefect cloud)

HI, I am a little confused as to how the agent works? How do the different agents differ? I have read the medium blog on deploying the [server] (https://medium.com/the-prefect-blog/prefect-server-101-deploying-to-google-cloud-platform-47354b16afe2). But from what I understand, we need the agent running for the scheduled tasks to run?

Hey all… found a busted link in the docs. On this, the link to the “sign up for our free Prefect Cloud Scheduler tier” goes to a 404

Hi, another newb question: What is the reasonable usage of

map()

? is using

map()

to handle a list of thousands (or millions?) of elements one by one reasonable? Should I micro-batch it into chunks? Does prefect start to choke if you have many many tasks in a flow? What about the Dashboard when you examine such a flow after it has run?

Hi all! @Laura Lorenz (she/her) is and is talking about how to set up your Prefect execution layer - feel free to drop by!

Silly question: How do I view & manage schedules if I'm not using the UI? I wanna deploy on a small server that's probably too small to run docker-compose.

Hello, Could someone help me with this error that I ran into when deploy flow into prefect-cloud?

File "build_and_deploy.py", line 48, in <module>
    env_vars=deployment_vars
  File "/Users/huizheng/.local/share/virtualenvs/data-platform-W017i1KI/lib/python3.7/site-packages/prefect/core/flow.py", line 1277, in deploy
    version_group_id=version_group_id,
  File "/Users/huizheng/.local/share/virtualenvs/data-platform-W017i1KI/lib/python3.7/site-packages/prefect/client/client.py", line 577, in deploy
    versionGroupId=version_group_id,
  File "/Users/huizheng/.local/share/virtualenvs/data-platform-W017i1KI/lib/python3.7/site-packages/prefect/client/client.py", line 222, in graphql
    raise ClientError(result["errors"])
prefect.utilities.exceptions.ClientError: [{'path': ['createFlowFromCompressedString'], 'message': '1 validation error for FlowSchema\nschedule -> clocks -> 0 -> parameter_defaults\n  field required (type=value_error.missing)', 'extensions': {'code': 'INTERNAL_SERVER_ERROR'}}]

I am using prefect version

0.7.3

(yes, it’s old, but have to stay with it for another month). the error was thrown at this line of code

dbt_flow.deploy(
        '{}'.format(env),
        python_dependencies=[
            'google-cloud-firestore', 'python-dotenv', 'google-cloud-storage', 'environs'],
        files={
            path.abspath('./prefect_cloud_deployment/{}/gcp-scheduler-key.json'.format(env)): '/home',
        },
        registry_url='<http://us.gcr.io/semios-dbt/scheduler|us.gcr.io/semios-dbt/scheduler>',
        env_vars=deployment_vars

The code before that line which might related to this error

# Flow auto-schedule
    start_time = datetime.now()
    start_time = start_time.replace(hour=(start_time.hour), minute=10, second=0, microsecond=0)
    hourly_schedule = Schedule(clocks=[IntervalClock(interval=timedelta(minutes=60), start_date=start_time)])
    dbt_flow.schedule = hourly_schedule

Hi Team, I’m trying Prefect following the guide doc, and found that there is a few seconds latency between triggering a flow to run via CLI “prefect run server --name hello-flow” and the agent runs the flow. Is that expected? Thanks.

Hi all, where can I find more information about how to run prefect from a container? The docs only show how to start a python interactive shell but how do I start the server or agents etc?

How to set

PREFECT__LOGGING__LOG_ATTRIBUTES

? I am trying to send a unique id while creating a flow run using python client. I want this uuid to be available in logs for tracking the request flow. However, so far I haven't had any success in correctly setting the extra log attributes. This is how i start the docker agent:

prefect agent start docker -e PREFECT__LOGGING__LOG_ATTRIBUTES="['uuid']"

I tried debugging and found that the log attributes are fetched as string here, instead of list.

Hi Prefect, I'm finding flows that have tasks that fail with 'No Heartbeat Detected' but the flow itself continues running (you can see in attached screenshots that heartbeat lost around 1:15AM but the flow continued running until manually marked as failed ~7 hours later). I believe previous Zombie Killer behavior is that the flow would have been immediately marked as failed. Is this an expected change to behavior?

Can Prefect UI be intefacesed with CloudWatch somehow? @josh @Jeremiah

Hi I thought I was being smart when modularizing the types of tasks, and having one module where I construct my flow:

from tasks.collect.games import *
from tasks.collect.streamers import *
from tasks.collect.streams import *
from tasks.enrich.game import *
from tasks.enrich.streamer import *
from tasks.enrich.stream import *
from tasks.store.game import *
from tasks.store.streamer import *
from tasks.store.stream import *
from tasks.util.common import *

with Flow("STRDATA POC") as strdata:
    collected_games = collect_games()
    enriched_games = enrich_game.map(collected_games)

    collected_streamers = collect_streamers()
    enriched_streamers = enrich_streamer.map(collected_streamers)

    collected_streams = collect_streams_per_game.map(enriched_games, unmapped(enriched_streamers))
    enriched_streams = enrich_stream.map(flatten(collected_streams))

    store_game.map(enriched_games)
    store_stream.map(enriched_streams)
    store_streamer.map(enriched_streamers)

The Flow runs OK when I run it standalone, but when I register it in my local prefect server, I can see the following error in the dashboard:

Failed to load and execute Flow's environment: ModuleNotFoundError("No module named 'tasks'")

it seems to be similar to an issue I found about not being able to submit flows to prefect cloud because of some peculiarity with pickle? https://github.com/PrefectHQ/prefect/issues/1742 But this was related to packaging the flow in a docker image, so I can't apply the solution to my case The layout of my project is the following:

deploy
|_
  prod
  |_
    register.py (contains flow.register)
flows
|_
  strdata_poc.py (contains flow definition - see above)
tasks
|_
  collect
  |_
    games.py
    streamers.py
    streams.py
  enrich
  |_
    ...

Hello Prefect Support, I am running into another error in my old-version preflect-flow local run, I am trying to run a flow locally and got the error below

[libprotobuf ERROR google/protobuf/descriptor_database.cc:58] File already exists in database:
[libprotobuf FATAL google/protobuf/descriptor.cc:1370] CHECK failed: GeneratedDatabase()->Add(encoded_file_descriptor, size):
libc++abi.dylib: terminating with uncaught exception of type google::protobuf::FatalException: CHECK failed: GeneratedDatabase()->Add(encoded_file_descriptor, size):
Abort trap: 6

The error is thrown when the code try to import

from main import dbt_flow

from .py file that defines the prefect flow. below are some libraries imported in the main.py file

from google.api_core.datetime_helpers import DatetimeWithNanoseconds
from prefect import task, Flow, Parameter, triggers, unmapped
from prefect.engine import signals, result_handlers
from prefect.schedules import IntervalSchedule
from datetime import timedelta, datetime, MAXYEAR
from google.cloud import firestore
from os import walk, path, getenv
from environs import Env
from collections.abc import Iterable
from itertools import chain
from copy import deepcopy
from time import sleep
import requests
import argparse
import json
import pytz

here is the pipfile

[packages]
prefect = "==0.7.3"
google-cloud-firestore = "==1.5.0"
python-dotenv = "==0.10.3"
environs = "==7.3.0"

[requires]
python_version = "3.7"

Hi everyone, I have been testing with concurrency tags and had a question about some behavior I am seeing with docker. It seems that after a task has been waiting (via concurrency tag limit) for 10 minutes, the prefect docker agent starts continuously spinning up new containers for that flow that is waiting to start. If I do a “docker ps” I could see 20 or more containers running for that flow. Any ideas on this?  My Setup:  Prefect Version: 0.10.6 Docker Engine Version: 19.03.6-ce Storage Type: Docker OS: Replicated the issue on Mac and Amazon Linux 2 Single prefect docker agent running 2 flows.

Hi folks, we used the GCSResultHandler back in 0.7.3. We see it writing files into GCS, but how best to decode them for debugging? I tried base64 decode, but that doesn't seem to work right.

Hi Team, I want to deploy the open source version of Prefect but authentication is required. Seems so far authentication is just supported in the cloud version. Do you have any good suggestions to achieve that?

Morning all! We’re seeing an issue with mapped tasks on Fargate, where we only get 2 mapped tasks running at a time. Doesn’t seem to matter what the tasks are, Prefect seems to cap the max concurrent tasks at 2. Any idea? We’re using Core 0.12.0, running on Fargate with DockerStorage and DaskExecutor And just uncovered another issue. While trying to boil down a sample flow to reproduce, I’m now hitting this problem on every run:

Unexpected error: KilledWorker('run_task-a6bb0ba3-e74f-4e26-8cc4-2bcc8f8c217c', <Worker '<tcp://127.0.0.1:41917>', name: 1, memory: 0, processing: 1>)

Doesn’t seem to matter what the flow is, getting this on all flows, even after restarting Prefect Server and the Fargate Agent..

Hello guys! Any idea why self hosted prefect server doesn't have the

flow_group

graphql query available? running 0.12.1

hi there - I’m having trouble telling my flow where to look for prefect server (the api). I’ve tried many different ways to set

prefect.config.api.url

(env variable

PREFECT__CONFIG__API__URL

and setting it in

config.toml

but when I call

.register()

I get the following error:

requests.exceptions.ConnectionError: HTTPConnectionPool(host='localhost', port=4200): Max retries exceeded with url: /graphql (Caused by NewConnectionError('<urllib3.connection.HTTPConnection object at 0x7fbcbd594e10>: Failed to establish a new connection: [Errno 111] Connection refused'))

indicating that it’s still trying

localhost

instead of what I’ve given it (which does not resolve to

localohost

) - any ideas?

I have a question about using

Parameter

in

ShellTask

According to the docs,

Parameter

is a type of

Task

, and I'm guessing it needs to get the method

run

called on it before it's usable. I'm guessing that this method is called when you pass the parameter to a function with the

@task

decorator? I have a shell task, which I'd like to pass an object (or a string) which is created using a

Parameter

object. Normally Shell tasks seem to be defined in the flow block. When I try to run a shell task, in a flow, with arguments that are dependent on a parameter, it fails with something like:

Traceback (most recent call last):
  File "/usr/lib/python3.8/runpy.py", line 193, in _run_module_as_main
    return _run_code(code, main_globals, None,
  File "/usr/lib/python3.8/runpy.py", line 86, in _run_code
    exec(code, run_globals)
  File "/home/macmenaminpe/code/prefect/pdb_flow/parameterized_flow.py", line 45, in <module>
    num_lines = s_task(command=f"wc -l {job.pdb_fp} > {job.job_dir}num_lines.txt")
  File "/home/macmenaminpe/.local/lib/python3.8/site-packages/prefect/tasks/core/function.py", line 68, in __getattr__
    raise AttributeError(f"'FunctionTask' object has no attribute {k}")
AttributeError: 'FunctionTask' object has no attribute pdb_fp

So, I guess the two questions are: • how do I "get at" the input val of a Parameter in a flow (or do I not do this, and always pass that param out to a

@task

and let it get handled there • how do I create a shellTask that takes args

Hi team, question about retries. Is there a good way to retry a task conditionally? Specifically, I want to retry a task on certain failures (e.g. db deadlocks). My thought was to use a state handler to detect the right situation (easy) but then how would I actually invoke the retry? Maybe return a new

Retrying

State object? Couldn’t find any doc examples of this and wanted to know if I missed anything obvious before spelunking. Thanks! 🙏

How do I add max_retries to a Task class? I modified the GCSDownloadAsFile class that inherits from GCSBaseTask class, and since this is a class, there is no @task decorator that I can add like

max_retries=3

to.

is this error expected?

TypeError: Object of type datetime is not JSON serializable

the relevant code is, outside the flow context:

@task(result=PrefectResult())
def assert_db_trigger(event_name: str, events_fetcher: DailyFlowEvents, point_date: datetime):
    events = events_fetcher.fetch_events()
    while event_name not in events.keys() or events[event_name] < point_date:
        print(f"No trigger, yet for {event_name}")
        sleep(5)
        events = events_fetcher.fetch_events()
    return point_date

and

point_datetime = datetime(year=point_date.year, month=point_date.month, day=point_date.day, hour=0, minute=0, second=0, microsecond=0, tzinfo=pytz.UTC)

and within the flow context:

task_wait_for_reference_data_validated = assert_db_trigger("reference_data_validated", daily_flow_events, point_datetime)

do all function/method arguments need to be JSON serializable?

I have an issue with File based flow storage. I followed the instructions (https://docs.prefect.io/core/idioms/file-based.html) and could not register the flow with

prefect register -f flows/my_flow.py

. I got an error

Error: no such option: -f

. When I register the flow with

flow.register()

it becomes available in UI but I am unable to run it, the tasks are pending and I can see this error

Failed to load and execute Flow's environment: ModuleNotFoundError("No module named 'github'")

. I am using

0.12.1

and Cloud. Any ideas?

I would like to run a given task with an array of dates:

@task
def generate_dates(start_date, n_days):
    return [start_date - datetime.timedelta(days=x) for x in range(n_days)]

@task
def do_something(updated_at):
    ## DO SOME STUFF WITH THE DATE

with Flow("My Flow") as flow:
    start_date = Parameter('start_date')
    n_days = Parameter('n_days')

    dates = generate_dates(start_date, n_days)

    for run_date in dates:
        do_something(run_date)

flow.run(dict(start_date=, n_days= 10))

In this case I don't want to use map for some technical reasons. I could just implement the loop inside of the task but I like having all the timing tracked by Prefect Cloud.

Hi lots, I know I have asked this question before in a previous thread. But I still have the problem that the (Core) Prefect Docker containers get a default name (in my case it is temp_xxx). I start the server application with "prefect server start". But I cannot find any parameters to influence the Docker Container name. Background is that I want to erase the Docker Container (by name) before each start for development reasons in the development environment. Thanks in advance.

Hi again, I am trying to bring up prefect using the below docker-compose file.

version: '3.7'

services:
  ad-ds-outlier-app:
    build: .
    command: >
      gunicorn -c "python:config.gunicorn"
        --reload "<http://myapp.app:create_app()|myapp.app:create_app()>"
    networks:
      - prefect-server
    volumes:
      - '.:/app'
      - './config/settings.py:/app/instance/settings.py'
    ports:
      - 12345:12345

  postgres:
    image: "postgres:11"
    ports:
      - "5432:5432"
    environment:
      POSTGRES_USER: prefect
      POSTGRES_PASSWORD: test-password
      POSTGRES_DB: prefect_server
    networks:
      - prefect-server
    restart: "always"
    command:
      - "postgres"
      # explicitly set max connections
      - "-c"
      - "max_connections=150"

  hasura:
    image: "hasura/graphql-engine:v1.1.0"
    ports:
      - "3000:3000"
    command: "graphql-engine serve"
    environment:
      HASURA_GRAPHQL_DATABASE_URL: "<postgresql://prefect:test-password@postgres:5432/prefect_server>"
      HASURA_GRAPHQL_ENABLE_CONSOLE: "true"
      HASURA_GRAPHQL_SERVER_PORT: "3000"
      HASURA_GRAPHQL_QUERY_PLAN_CACHE_SIZE: 100
    networks:
      - prefect-server
    restart: "always"
    depends_on:
      - postgres

  graphql:
    image: "prefecthq/server:latest"
    ports:
      - "4201:4201"
    command: bash -c "prefect-server database upgrade -y && python src/prefect_server/services/graphql/server.py"
    environment:
      PREFECT_SERVER_DB_CMD: "echo 'DATABASE MIGRATIONS SKIPPED'"
      PREFECT_SERVER__DATABASE__CONNECTION_URL: "<postgresql://prefect:test-password@postgres:5432/prefect_server>"
      PREFECT_SERVER__HASURA__ADMIN_SECRET: hasura-secret-admin-secret
      PREFECT_SERVER__HASURA__HOST: hasura
    networks:
      - prefect-server
    restart: "always"
    depends_on:
      - hasura

  scheduler:
    image: "prefecthq/server:latest"
    command: "python src/prefect_server/services/scheduler/scheduler.py"
    environment:
      PREFECT_SERVER__HASURA__ADMIN_SECRET: hasura-secret-admin-secret
      PREFECT_SERVER__HASURA__HOST: hasura
    networks:
      - prefect-server
    restart: "always"
    depends_on:
      - graphql

  apollo:
    image: "prefecthq/apollo:latest"
    ports:
      - "4200:4200"
    command: "npm run serve"
    environment:
      HASURA_API_URL: <http://hasura:3000/v1alpha1/graphql>
      PREFECT_API_URL: <http://graphql:4201/graphql/>
      PREFECT_API_HEALTH_URL: <http://graphql:4201/health>
      PREFECT_SERVER__TELEMETRY__ENABLED: "true"
    networks:
      - prefect-server
    restart: "always"
    depends_on:
      - graphql

  ui:
    image: "prefecthq/ui:latest"
    ports:
      - "8080:8080"
    command: "/intercept.sh"
    environment:
      PREFECT_SERVER__GRAPHQL_URL: <http://localhost:4200/graphql>
    networks:
      - prefect-server
    restart: "always"
    depends_on:
      - apollo

  db:
    image: 'ellerbrock/alpine-mongodb'
    environment:
      SERVICE_HOST: 0.0.0.0
    volumes:
      - ./db/mongodb/db:/data/db
      - ./entrypoint.sh:/entrypoint.sh
    networks:
      - prefect-server
    ports:
      - 27017:27017

volumes:
  db:

networks:
  prefect-server:
    name: prefect-server

The only additions to the original prefect provided docker-compose are a service and db. When I perform a

flow.register

, I get an error which states,

'project_name' is a required field when using prefect cloud

. I am running this locally and do not know where do I mention to use backend instead of cloud by default. The normal way is on the host I can simply run

prefect backend server; prefect server start

, but since I am using a custom docker-compose file, not sure how to get around this. Any help? 😕

Hey guys, question around getting Parameter values and creating the execution environment for a flow. I need to use an input Parameter as a config value for creating a number of Dask workers, but it seems like I can’t get at the value till runtime. Sample code below. Any thoughts on how to get around it?

def main(registry, name, tag, base_image, env_vars, python_dependencies, execution):
    worker_count = 1

    with Flow("Parallell Orchestrator") as flow:
        bucket_name = Parameter("bucket_name", default="test-bucket")
        data_folder = Parameter("data_folder", default="data/parallellTest")
        collector_flow = Parameter("collector_flow", default="Parallell Test")
        partition_count = Parameter("partition_count", default=1)
        input_file = Parameter("input_file", default="data/parallellTest/output/results.jsonl")

        output_folder = data_folder + "/output"

        ############# THE MAGIC LINE #############
        worker_count = partition_count
        ##########################################
 
        # Run Partition Flow
        partitions = partition_json_data(bucket_name, input_file, data_folder, partition_count)

        # Run Parallell
        final_states = trigger_subflow.map(partitions, unmapped(bucket_name), unmapped(collector_flow))

        #  Run Collation Flow
        collate_results(bucket_name, output_folder, final_states)

    flow.storage = Docker(registry_url=registry, image_name=name, image_tag=tag, env_vars=env_vars,
                          base_image=base_image, python_dependencies=python_dependencies, prefect_version="master")

    flow.environment = LocalEnvironment(executor=DaskExecutor(n_workers=worker_count))