Hello! Is there a reason the credentials implementation in prefect-github is a Python dataclass and not a Prefect block (as it is in prefect-aws.credentials, for instance)?

Hi, a simple question for Prefect1.0, I define a function (but not decorated with @task decorator). This function does a third party API call to return a list of dictionary. Use case is this function will be called in

with Flow

block and the list it returns will be used in

create_flow_run.map(parameters=...)

, My question is when we deployed the flow (it runs every midnight), will the function we defined be called each time when Flow is scheduled to run?

👋 peeking into the future a little bit for prefect (also apologies if this already exists and i missed it), do you think we might be able to specify different caching mediums (e.g. cache using a distributed cache like redis)?

Running a Prefect 2.0 Hello World workflow with a GKE cluster from the image

prefecthq/prefect:2.0.2-python3.10

and when trying to access GCS it appears that

gcsfs

is not installed and am getting the following error from my flow, should I build a custom image that installs

gcsfs

?

Flow could not be retrieved from deployment.
Traceback (most recent call last):
  File "/usr/local/lib/python3.10/site-packages/fsspec/registry.py", line 232, in get_filesystem_class
    register_implementation(protocol, _import_class(bit["class"]))
  File "/usr/local/lib/python3.10/site-packages/fsspec/registry.py", line 255, in _import_class
    mod = importlib.import_module(mod)
  File "/usr/local/lib/python3.10/importlib/__init__.py", line 126, in import_module
    return _bootstrap._gcd_import(name[level:], package, level)
  File "<frozen importlib._bootstrap>", line 1050, in _gcd_import
  File "<frozen importlib._bootstrap>", line 1027, in _find_and_load
  File "<frozen importlib._bootstrap>", line 1004, in _find_and_load_unlocked
ModuleNotFoundError: No module named 'gcsfs'

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

Traceback (most recent call last):
  File "/usr/local/lib/python3.10/site-packages/prefect/filesystems.py", line 312, in filesystem
    self._filesystem = fsspec.filesystem(scheme, **self.settings)
  File "/usr/local/lib/python3.10/site-packages/fsspec/registry.py", line 265, in filesystem
    cls = get_filesystem_class(protocol)
  File "/usr/local/lib/python3.10/site-packages/fsspec/registry.py", line 234, in get_filesystem_class
    raise ImportError(bit["err"]) from e
ImportError: Please install gcsfs to access Google Storage

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

Traceback (most recent call last):
  File "/usr/local/lib/python3.10/site-packages/prefect/engine.py", line 247, in retrieve_flow_then_begin_flow_run
    flow = await load_flow_from_flow_run(flow_run, client=client)
  File "/usr/local/lib/python3.10/site-packages/prefect/client.py", line 104, in with_injected_client
    return await fn(*args, **kwargs)
  File "/usr/local/lib/python3.10/site-packages/prefect/deployments.py", line 47, in load_flow_from_flow_run
    await storage_block.get_directory(from_path=None, local_path=".")
  File "/usr/local/lib/python3.10/site-packages/prefect/filesystems.py", line 454, in get_directory
    return await self.filesystem.get_directory(
  File "/usr/local/lib/python3.10/site-packages/prefect/filesystems.py", line 251, in get_directory
    return self.filesystem.get(from_path, local_path, recursive=True)
  File "/usr/local/lib/python3.10/site-packages/prefect/filesystems.py", line 315, in filesystem
    raise RuntimeError(
RuntimeError: File system created with scheme 'gcs' from base path 'gcs://<bucket>/deployments' could not be created. You are likely missing a Python module required to use the given storage protocol.

How do you set up a parameter in a parent flow to be passed after modification into a child flow? I have problems in the using of the parameter. Ive tried...

DOW = Parameter("DOW", default="Monday")

FlowExecution = create_flow_run(

flow_name=FlowName,

project_name=ProjectName,

task_args=dict(name="Flow Execution",trigger=all_successful),

parameters={"StringA":"The Day of the week is "+DOW+"."}

)

AND

DOW = Parameter("DOW", default="Monday")

StringAField = "The Day of the week is "+DOW+"."

FlowExecution = create_flow_run(

flow_name=FlowName,

project_name=ProjectName,

task_args=dict(name="Flow Execution",trigger=all_successful),

parameters={"StringA":StringAField}

)

Hello! Has anyone been using Prefect 2.0 with Dask managed with Coiled? I am having issues with distributing a dataframe across multiple Dask workers within the same cluster. The cluster is deployed with 4 workers but the dataframe is only being loaded in 1 of the workers, leaving the 3 others idle. When I run the same script outside of a Prefect task & flow, it runs as expected - the dataframe gets distributed across the cluster for later use. I also tested another scenario where the cluster is deployed with only 1 worker, but with autoscaling enabled by passing adapt_kwargs={"maximum": 10} to DaskTaskRunner. In this scenario, it seems like the autoscaling request is ignored as the cluster remains of size 1. Lastly, is there a way to return the standard output of a print() statement executed on a remote dask cluster back to the Prefect logger?

from prefect import flow, task
from prefect_dask.task_runners import DaskTaskRunner
import dask

coiled_executor = DaskTaskRunner(
	cluster_class="coiled.Cluster",
	cluster_kwargs={
		"n_workers" : 4,
		"software": "ttibi-dev",
		"shutdown_on_close": True,
		"worker_vm_types":["r6a.large"]
	},
	adapt_kwargs={"maximum": 10}
)


@task
def some_data_manipulation():
	df = dask.datasets.timeseries(
		"2000", "2020", partition_freq="2w"
	).persist()
	df.groupby("name").aggregate({"x": "sum", "y": "max"}).compute()
	return df

@flow(task_runner=coiled_executor)
def test_flow():
	some_data_manipulation.submit()

if __name__ == "__main__":	
	test_flow()

Me again Why if I update permissions of a role to only be 1 thing, it still has access to things I did not specify ? Does it mean that those "Admin read values" are always ON for all users ?

Last chance to submit your Code Contest submissions! Please post them in #show-us-what-you-got by midnight Eastern time tonight!

Team, My python program saves a csv file as an output in the same directory as of program, when I am trying to run this pipeline through prefect:2.0 cloud, I am getting this error. How I can get rid of it? The storage is configured by prefect itself, and when I am trying to specify storage block, deployment.yaml file is not being created. Also, Any working prefect 2.0 example which utilizes remote storage?

Hi all, I'm wondering if we can manually kill a task_runner?

Hi everyone, I'm trying to create a schedule task in a newest version but it seem didn't work. Anyone can help me because I was reading docs so many times but I couldn't find a way.

Version: 2.0.3

from prefect.schedules import IntervalSchedule

ModuleNotFoundError: No module named 'prefect.schedules'

Another thing regarding Schedules. According to this doc page:

Deployments are changing, and along with them the way you specify a schedule on a deployment. Stay tuned for updated guidance.

Currently it's not very clear to me how to set a schedule for my flows. Is it set in the flow code, or have to be set via the command line?

Hi everyone! I'm interested in open data catalog services. I'm trying open metadata and amundsen. I've seen that both only support Apache Airflow for ETLs orchestration. Do you know if Prefect is working with open metadata, amundsen, or other similar products? Thank you!

Hello, I can't login on prefect cloud from my work computer using the CLI :

prefect cloud login -k xxxxxxxxxxxxxxxx

. I'm getting the following error :

ssl.SSLCertVerificationError: [SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: unable to get local issuer certificate (_ssl.c:997)

. Is there a way to solve this ? I am using a windows machine and tried setting CURL_CA_BUNDLE=’’, PYTHONHTTPSVERIFY=‘false’ , but it didn't work

I am not sure whether I am the only one facing the issue, but how can u deploy prefect orion permanently without using the cloud solution? In 1.0 it was fairly easy to adjust the docker-compose file. Now, I can run the orion server on our inhouse-cluster or a cloud-vm, but since I am not able to docker-compose it (somehow the communication between the db and orion is prevented although they are in the same network) , whenever the command line closes, where the server is start from, the server shuts down. Same for the agent. Maybe I completely miss something here?

Hi all, I'm trying to deploy a flow in prefect 2.0 but keep encountering this "flow not found" error, even when the flow is clearly there. Is there anything in the .py file that can possibly prevent the flow from being detected?

Hi, I am trying to run deployment on K8s based on these docs: https://docs.prefect.io/tutorials/_kubernetes-flow-runner/ But when I try to run the command:

prefect deployment create ./kubernetes-deployment.py

I get this error:

No such command 'create'.

My prefect version is 2.0.0 Anyone know what might be the cause?

Hi I’m trying to use Blocks for using secrets in my ETL. However, I keep on running into errors when trying to load them:

---------------------------------------------------------------------------
AttributeError                            Traceback (most recent call last)
Input In [1], in <cell line: 6>()
      3 secret_block = Secret.load("etlharvestqaspassword")
      5 # Access the stored secret
----> 6 secret_block.get()

AttributeError: 'coroutine' object has no attribute 'get'

I defined the block within the Orion UI, because when I tried to define it via code in a simple script (as suggested here), I get the following kind of error:

prefect.exceptions.PrefectHTTPStatusError: Client error '422 Unprocessable Entity' for url '<http://ephemeral-orion/api/block_documents/>'
Response: {'exception_message': 'Invalid request received.', 'exception_detail': [{'loc': ['body', 'name'], 'msg': 'name must only contain lowercase letters, numbers, and dashes', 'type': 'value_error'}, {'loc': ['body', '__root__'], 'msg': 'Names must be provided for block documents.', 'type': 'value_error'}], 'request_body': {'name': 'test2_password', 'data': {'value': 'test2'}, 'block_schema_id': '8019abd6-409a-4f91-9367-bc8343c31763', 'block_type_id': '29fb0ec8-f7e9-4527-984c-48f8675f2bc4', 'is_anonymous': False}}
For more information check: <https://httpstatuses.com/422>

I’m mostly using Prefect within a jupyter notebook and from within a virtualenv. Thanks in advance for anyone who could point me to what’s going on 🙂

Does anyone know of a workaround to run a task from a task in Prefect 2? I've found some examples of people using .fn but it doesn't appear to work in my case. Specifically, I'd prefer to run trigger_dbt_cli_command from the prefect-dbt package from one of my tasks rather than having to create a flow to run it. I can't change the definition of trigger_dbt_cli_command, since the package itself defines it as a task. I think my problem might be related to the task being async, it seems to finish the execution instantly and the returned type is "None". As long as I'm not using fn(), Prefect will patiently wait for a result, but then I'm forced to call the task from a flow. Edit: If anyone thinks I'm going about this the wrong way, I'd be happy to hear that too. Maybe I just need to settle on always calling trigger_dbt_cli_command from a flow. 🙂

Hello community 🦜, Suppose we have 1 queue and 5 agents working with this queue. We can set concurrency-limit for queue. Can we set concurrency limit for each agent to be sure that all agents work with 1 flow only at the same time ? 🤷‍♂️

Hi Trying to create a deployment per the instructions on https://docs.prefect.io/tutorials/deployments/, I was a bit confused by the sentence “and the name of the entrypoint flow function, separated by a colon.“. I thought it was possible to specify the name of the flow (given inside the decorator) after the colon, whereas you need to specify the name of the function that contains the main flow. Perhaps the sentence should read something like “and the name of the function containing the main flow, separated by a colon.” 🙂

can someone help with his error

Hey, since yesterday, i try to run an etl flow with prefect 1 running on a eks fargate cluster. I tried it 2 times and it always exits with a k8 error 255 after about an hour of runtime. I don't really have any idea on why it is happenning as i'm fairly novice with k8's and aws

Hi all, I'm trying to use

prefect-email

to send some dummy test email (I know there's prefect notifications features) but I want test out

prefect-email

as well, got this error everytime, hard coded password for dummy test, can't be a wrong password 🤔

smtplib.SMTPAuthenticationError: (535, b'5.7.8 Username and Password not accepted. Learn more at\n5.7.8  <https://support.google.com/mail/?p=BadCredentials> d6-20020a170903230600b0016efc27ca98sm3023696plh.169 - gsmtp'

Thank you

Hi, a simple question 🙂 is there a way to register the flow one by one in the correct order?

hey all, I want to implement custom

Result

like implementation for 2.0, but I can’t find anything in the docs. Closest thing I am seeing is

FileSystems

but I don’t see example usage of passing data between tasks using a custom persistence layer

Hi- what’s the best practice to store a file in local storage and process in the same task? I tried to write a file and send it, but it only works locally. When I deployed it to run on the k8s cluster, it always report this error of

file not found

. So I cannot store it in

/tmp

and process later in this task?

Hi, trying to upload files generated from my extraction step to GCS and found the examples in

prefect-gcp

to do it, but when I combine this with block information it seems like the info is not in the correct format.

gcs_block = GCS.load("gcs-dev")

@flow()
def example_cloud_storage_upload_blob_from_file_flow():
    gcp_credentials = GcpCredentials(service_account_info=gcs_block.service_account_info)
    test_upload_file = "test_upload.txt"
    blob = cloud_storage_upload_blob_from_file(test_upload_path, gcs_block.bucket_path, "test_upload.txt", gcp_credentials)
    return blob

Hi All - I have a question about the expected behavior of flows when using task mapping. When I use "apply_map" to map ~1500 tasks what I see in the UI is only a set of 8 "Constant[list][x]" tasks as in the screenshot below. I originally thought perhaps this was somehow tied to use of DaskExecutor but the number of "Constant" tasks is not the same as the number of dask workers I have running. In addition, each Constant task shows the same x/1500 task running, rather than a subset. Based on runtime, it seems plausible that each Constant task is processing all the data. Still doing testing to confirm this but it felt like I was doing something wrong that would be clear to more experienced users. I've included some simple code to represent my implementation below as well. Thanks for the help!

@task
def actual_task(arg1, arg2, arg3):
   #does stuff

task_list = [
    (1, "a", "b"),
    (2, "c", "d"),
    (3, "e", "f"),
    (4, "g", "h"),
]

def task_map(task):
    return actual_task(task[0], task[1], task[2])

with Flow("my_flow") as flow:
    task_results = apply_map(task_map, task_list)