Hi community, i am new to prefect and trying to wrap my head around the framework. I have a simple ETL flow that i need to implement reading a large collection from mongodb, transforming it, and saving back the result to a mysql table. The collection is too big to fit in memory so i need to be able to do some sort of batching but i am struggling to understand how. i was thinking of fixing a LIMIT constant, calculating the count of the collection and producing map tasks over a

range(0,count,LIMIT)

pseudo code:

@task 
def count_collection_mongo(collectionName):
    # calculate number of rows in the collection and return it
    return count

@task
def load_data(collectionName, skip, limit):
    # data = db.collectionName.find().skip(skip).limit(limit)
    return data

@task
def transform(data):
    print(data.get('name'))
    return data

@task
def calculate_iterations_data(count_result):
    res = range(0, count_result, LIMIT)
    return res


with Flow("ETL flow for mycollection") as flow:
    result_count = count_collection_mongo("mycollection")
    iterations = calculate_iterations_data(result_count)
    data = load_data.map(iterations)
    transform.map(flatten(data))

flow.run()

is this the correct way to tackle the issue? How i can make sure that memory will not blow up as each of the load_data will save its result in memory, how can i control the number of parallel transform that can run ?

My flow worked fine until I add a GCP Data Catalog and Big Query piece in my code which is helping me fetch some column names from the Data Catalog and update some data in Big Query. Once I added the Data Catalog Piece and Big Query client.

from google.cloud import datacatalog_v1
from google.cloud import bigquery

client = bigquery.Client()
datacatalog_client = datacatalog_v1.DataCatalogClient()

I get an error saying that: Clients have non-trivial state that is local and unpickleable Can someone tell me what does this mean? I'm assuming this has something to do with GCP. I'm using a local service account json file and also importing pubsub modules, I didn't have any issue then.

I have my flows all converted from prefect core to running in prefect server and coming back to getting some parallelism I thought that this:

from time import sleep

import prefect
from prefect import Flow, task
from prefect.executors import LocalDaskExecutor, DaskExecutor
from prefect.storage import Docker


@task
def slow_task(item, sleep_time=4):
    logger = prefect.context.get('logger')
    <http://logger.info|logger.info>(f"==== IN TASK item {item} sleeping {sleep_time}")
    sleep(sleep_time)
    <http://logger.info|logger.info>(f"## Awake {item}")
    return item


@prefect.task
def produce_range(rangelen):
    return range(rangelen)


with Flow("parallel") as flow:
    nrange = produce_range(6)
    numbers = slow_task.map(item=nrange)

flow.executor = LocalDaskExecutor(workers=10, scheduler="processes")
#flow.executor = DaskExecutor()

if 'local_script_path' in prefect.context:
    flow.storage = Docker(registry_url='<http://old7.mianos.com:5000|old7.mianos.com:5000>')


if __name__ == '__main__':
    flow.run()

Would start 10 processes inside the docker container running, returning after a few seconds but it executes each task serially unless I run it without any arguments to LocalDaskExecutor. Under k8s DaskExecutor seems to start a few threads but nowhere near enough to what I want (30+). I feel I am missing something.

Hi All, is anyone using Module storage option?

Hey everyone, is there a good way to access flow-run metadata within a flow context? For example, a task that grabs the current flow-run-id or even an upstream task's duration. Or will pulling metadata need to be done in a separate flow?

hi guys, we had a docker agent fail over the weekend, so there were like 100 late scheduled runs as soon as i restarted the agent, it immediately queued and started all 100 runs these are pretty expensive tasks, so firing them all up at once immediately ground the server to a halt, couldn't ssh in to try killing things so my questions coming out of this are: • How do you configure the scheduled services to just give up if they get missed? We run the task every hour, so it's fine to just run it the next hour instead. Having it give up after, say, 30 minutes of waiting, would be very elegant, but just failing immediately is perfectly fine • How do you mass cancel scheduled jobs that have missed their time? • How do you mass cancel running jobs? I manually clicked open every job and pressed cancel until the system starting responding enough to ssh and I could go in and start killing things via the docker cli. With how unresponsive the UI is in general, this was pretty painful. • Is there some way to kill the agent to stop everything? This is the idea of some kind of breaker switch -- prefect just started going crazy on this server, shut it all down so we're back to a working state, and then we can sort it out afterwards. • After all the jobs had been marked as Cancelled or Failed, there were still containers running on the server. Are these just all the Failed containers that lost their heartbeat? Is there a nicer way to clean them up, or are Failed containers just abandoned by Prefect at that point, and the answer is to just go in and manually remove them like I did? Does that mean that any Failed tasks should always be manually reviewed for any hanging containers that need to be stopped and removed? • How do you configure a limit for the agent, so it can't get bad enough to make the server inaccessible to repair? I know based on tags that this agent is only going to run jobs of a certain type, so I could set the limit to, say, 5, and it'd just have 5 running 95 queued, and then I'd just need to mass-cancel those queued jobs.

Does the cost calculator at https://www.prefect.io/pricing/ work for anyone? I can't enter numbers higher than 9999

Also it would be great if the blue keywords would show detailed description of the feature on mouse over or link to a page with it

I just ran one of my new flows in production, scripts run on 3500 machines

hey everyone, is it possible to capture default ‘logging’ logs on prefect run? we’re using many libraries that print valuable logs but don’t have the context of the prefect logger. we want instead of pushing forward the logger object to be able to use the global python logging object

Hi everyone. I am working with a data science team that wants to orchestrate some workflows. They already have scripts that run inside of a docker container. I was thinking that the easiest way to add these to the flow would be to use the Docker Tasks (

CreateContainer

,

StartContainer

, etc.). However, it seems that the logs would need to be retrieved after the container finishes running with

GetContainerLogs

. Is this correct? If so, this is less than ideal for our use case given that these are long-running processes we'd want to see the logs in real-time. So far, I've thought about a couple of alternatives: 1. Modify the Docker Tasks to use the

logs

method with

stream=True

(I haven't tested this yet, but the docs suggest this could work) 2. Add prefect to their docker image and create a flow that runs inside of the image Do you see another option? What would you recommend? Thanks!

Is it possible to change the start time of a scheduled workflow?

Is there any way to define which task executes first and second task has to be executed only after first task is executed Both of these tasks loads same data from a loader function

Hello, I am using KubernetesRun for the flow run configs.. I am trying to specify a

service_account_name

, but it seems the pods are still using the default service account to the namespace.. my agent is also deployed with an undefined

service_account_name

.. any thoughts?

flow.run_configs = KubernetesRun(
    image_pull_secrets=["xxx"],
    job_template=config.KUBE_JOB_TEMPLATE,
    cpu_limit=config.KUBE_CPU_LIMIT,
    cpu_request=config.KUBE_CPU_REQUEST,
    memory_limit=config.KUBE_MEMORY_LIMIT,
    memory_request=config.KUBE_MEMORY_REQUEST,
    service_account_name="my-service-account",
)

Hello. I’m attempting to use the DBTShellTask to execute DBT tasks and stream the DBTShellTask output to the Prefect flow logs. This is working locally when I start a local agent and register and run the flow, but the same code is not streaming the output when executing inside a docker container. Any suggestions?

@task
def run_DBT_task():

    task = DbtShellTask(
        profile_name='snowflake',
        environment='sandbox',
        profiles_dir='.',
        stream_output=True,
        return_all=True
    )
    
    task.run(command="cd ~/src/prefect && dbt run")

with Flow(name="Local Test") as flow:
    run_DBT_task()

flowID = flow.register(project_name="51", labels=["localtest"])

client.create_flow_run(flow_id = flowID)

There is something that is a bit unclear to me regarding the prefect loggers. What problem does it solve? What is the difference between:

@task
def mytask():
    logger = prefect.context.get('logger')
    <http://logger.info|logger.info>('hello')

and

logger = logging.getLogger(__name__)

@task
def mytask():
    <http://logger.info|logger.info>('hello')

i.e. What is the point of using a prefect-provided logger instead of regular python logging loggers?

is it just a way to have per-task loggers? is there no difference? Basically i'm running into issues of duplicated log messages with prefect loggers, which is kind of annoying. I'm wondering if the whole concept of using these loggers instead of standard python loggers is actually useful

and the doc is quite unclear as to why these prefect loggers exist, it just mentions that they exist so I assumed it's better to use them, but I have no idea what they actually do

Would there be a recommended way for a client to block waiting for a flow to complete? I am sending the task request with a combo of graphql and the client functions (like prefect flow run) but I would like the endpoint to block for a while for shorter tasks. Just looking for ideas. (I have puslar integrated so I might simply subscribe and have a task in the end of the flow publish, unblocking the client).

Hey folks! Does anyone have an example repo they could share that's handling the deployment of a Prefect Agent with AKS? Currently starting off with spinning up a cluster with Terraform, but my k8s skills are sub-par so some useful starters would be handy. For reference i've CDK'd up a Prefect Agent & respective cluster in ECS on AWS before, but that wasn't involving k8s.

Hi folks! I'm wondering if there's a way to code a Task that would be mapped, but initialized only once. I want to map a rollbar task onto a result (list) from a previous task, but ideally I would like to call the rollbar.init() only once. Anyone know how I could do that?

I'm running into an issue where I cannot build my flow (using docker storage). I am running:

prefect build -p /home/circleci/project/data_flows/flows/vaccine_cdc.py

(log in thread)

sample.py

Hi guys! Happy to join your community! Looking for ways to put my 'Data mess' in order

I have a single flow where I want to limit concurrency to one flow run at a time for business logic reasons. I see that in Prefect Cloud I can limit concurrency at a Flow level using labels. Currently we use labels to tag flows by team (business-intelligence, data-engineering, machine learning, etc) and by environment (uat, prod) for routing to the appropriate agent. Is there any other way of limiting flow concurrency based on something other labels (e.g. flow name and project)? I can make it work, but I'm just not crazy about the idea of adding another label that would specify flow name.

Is it possible to drop the index with the PandasSerializer with serialize_kwargs?

Where can I find

~/.prefect/config.toml

in Windows 10

For S3Result.write(), Returns: • `Result`: a new Result instance with the appropriately formatted S3 URI I wasn't able to return the formatted S3 URI, only the S3 Result itself. Any pointers? I'm doing:

data = results.write(df, **prefect.context)
print(data)

Hello everyone, for CI/CD purposes, we need to call prefect from Jenkins. There is no Restful API in prefect, so we need to use GraphQL. Is there any good example to share for Jenkins / Prefect integration?