https://prefect.io
Join Slack
I don't understand what's the good practice to work with block during the flow. I have a block that ...
# ask-community
f
I don't understand what's the good practice to work with block during the flow. I have a block that I want to pass in some tasks of my flow. When I run it locally, it works. But when I run a deployment on the cloud, each task get a dict with this format: 
{'$ref': {'block_document_id': '779148c3-8067-496d-9c03-0ae3f81de6ca'}}
instead of the block instance. How can I pass a block instance in a task of my flow?
n
hi @Florent VanDeMoortele - where are you seeing this? 
{'$ref': {'block_document_id': '779148c3-8067-496d-9c03-0ae3f81de6ca'}}
and if you don't mind, how are you passing your block to your tasks?
f
I log it
I pass my block as an argument
I select my block with the Prefect UI, then I run the flow
Here is the schema I use :
Copy code
@task
def task_1(bucket: GcsBucket, **kwargs):
   # Here I log and I obtain the previous dict instead of block instance
   # Retrieve my file from GcsBucket and some others functions

@flow
def subflow(bucket: GcsBucket, **kwargs):
   # Here I log and I obtain the previous dict instead of block instance
   task_1.map(arg1, unmapped(bucket))
   task_2 (no need of Block)
   task_3.map(arg2, unmapped(bucket))

@flow
def flow(bucket: GcsBucket, **kwargs): # Here I select block instance from the UI
   subflow(bucket, **kwargs)
   other_subflow
@Nate I try also without the flow and subflow but just with a simple flow, it's the same result
n
hmm im having a hard time reproducing that. you can share what your remote infra is like? or if you’re encountering this with specific block types?
f
I have this problem with custom blocks and also from collections (like prefect_gcs here with GcsBucket block)
👍 1
For my deployment I use docker, my work pool is a K8s jobs
Locally, it works. I tried to ask Marvin (https://prefect-community.slack.com/archives/C04DZJC94DC/p1692201060982099?thread_ts=1692187323.239879&cid=C04DZJC94DC ), and I think the problem is that on the cloud deployment, arguments are serialized to be pass over task
But impossible to deserialize back (SQL query to retrieve block name seems to be overkill)
If blocks aren't design like that, maybe do you have an other suggestion to share blocks over tasks and flows?
load the same block in each task seems to be unoptimized for me
n
You see, when you pass a block instance to a task, Prefect serializes the block instance into a dictionary.
Marvin is incorrect here - you're free to pass types like Blocks to tasks (and thats not how blocks look when JSON serialized) for example, this works fine both locally and remotely
Copy code
from prefect import flow, task, unmapped
from prefect.blocks.core import Block

class MyBlock(Block):
    custom_field: str = "default value"
    
# MyBlock().save("test")

@task
def accept_foo_and_block(foo: str, block: Block):
    print(f"Received foo {foo!r} of type {type(foo)}")
    print(f"Received block {block!r} of type {type(block)}")
    
@flow(log_prints=True)
def some_flow():
    accept_foo_and_block.map(
        foo=["a", "b"], block=unmapped(MyBlock.load("test"))
    )

if __name__ == "__main__":
    some_flow()
it seems there must be something going on with the block types that are giving you the 
'$ref'
f
Thank you @Nate. I try with your example. When I load my block into the flow: it's okay. But, here block is an argument of my flow (to allow to chose into the Prefect UI). If I try with:
Copy code
@flow(log_prints=True)
def some_flow(block: MyBlock):
    accept_foo_and_block.map(
        foo=["a", "b"], block=unmapped(block)
    )
It doesn't work. I think the problem append when Prefect UI transmits arguments to flow. Do you have any idea to resolve this problem?
Found it. On the Prefect UI, when I select my block in the custom form and then I switch to JSON form, input looks like :
Copy code
{
  "gcs_bucket": {
    "$ref": {
      "block_document_id": "779148c3-8067-496d-9c03-0ae3f81de6ca"
    }
  }
}
I tried multiple things, without success. The only way is to pass block name in flow arg instead of block instance. Maybe do you have an other suggestion?
n
hmm im not sure i understand your problem. if i have a flow like this
Copy code
from prefect import flow
from prefect_gcp import GcsBucket

@flow(log_prints=True)
def foo(bucket: GcsBucket):
    print(bucket)

if __name__ == "__main__":
    foo()
and deploy it, I can use the form view to select / create a block, or go to the JSON tab to specify a JSON representation of the block are you not seeing this behaviour?
f
Yes, exactly. When I use the form, I select the block and when my flow run I obtain a dict representation with $ref instead of block instance
And from this dictionary, I can't retrieve a block instance (and access to child method, and so on)
If you try to run your example by selecting a GcsBucket block, it works for you? You can print the block instance? Or you have the representation with $ref ?
I work with prefect 2.11.3, the latest
n
interesting - I see the behaviour you're describing now. certainly seems like a bug
short term fix is to pass a block name as you mention, but we should open an issue here and figure out what exactly is going on - do you feel comfortable doing that? otherwise I can
f
This morning, I found this PR: https://github.com/PrefectHQ/prefect-ui-library/pull/1557 It's possibly a similar issue
For now, I switch my flows to use block name instead of the block instances. Thanks @Nate 🙏
Thank you for your help and for your time ! If it's ok for you, I let you open the issue, then I'll subscribe to it to follow changes.
n
sounds good!
f
Hi @Nate , I can't retrieve the Github issue about the block bug in UI. Could you please send me the link? So I can subscribe to it and follow this issue. Thank you !
19 Views
Open in Slack
PreviousNext
Powered by

Prefect Community

Bring your towel and join one of the fastest growing data communities. Welcome to our second-generation open source orchestration platform, a completely rethought approach to dataflow automation.

Powered by