Hey there, maybe you can help me out here with some Prefect code and the rendered schematics. Context: I’m developing a flow that creates and submits simple spark apps to EMR. To my surprise there are no built-in tasks for EMR (docs). Nonetheless, I am trying to use

boto3

and

awswrangler

APIs (as I’ve seen others users asking for it on the community channel). My current issue, is on the lack of understanding on how the Flows are rendered on the Schematic view. It’s related to tasks dependencies and inter-task communication (passing values).

When you say improve the flow schematic, do you mean some sort of drill-down functionality?

So I am not sure this can be improved because you are using that

cluster_id

, which is a task, in downstream tasks so it gets connected. I think the only way this schematic can be improved is if

cluster_id

is hardcoded, which loses a lot of flexibility. I believe this can be done in Orion because you can materialize tasks as Python objects by calling

.wait().result()

so you can materialize the

cluster_id

task and then pass it downstream as a normal Python object

Another thing that could improve your flow in general would be to leverage a resource manager for the boto3 client that you are invoking within the flow. Instead of passing this HTTP client between tasks, you could pass it in a way that would be safer and would provide you even more visibility into it within your flow, This blog post has an example

thank you both @Kevin Kho and @Anna Geller! 1. that

.wait().result()

would probably very useful! Another thing I’ve noted was that if I’m grabbing a value from a task result, let’s say my task it’s returning an array and I want to use it as arguments to multiple tasks ( e.g.

task2(arg1=task1['result1']

) , the graph renders new boxes/nodes for that intermediate results. 2. Didn’t know about that

resourse manager

feature! I’ll look into it, it seems very appropiate for my use case. Thanks for sharing. 3. I don’t wanna use a 24/7 cluster, I delegate the creation and destruction to the orchestration tool: a create cluster, run steps, terminate approach. I don’t want to submit all the steps at the creation time to have control of the submission, data dependencies between my spark apps and not rely on EMR to do so. I’ll have a timeout for EMR to kill itself as part of cluster configs (EMR > 6.0.0). 4. I’m not using awsdatawrangler yet because the method to create cluster uses instance fleets and not instance groups. For this example I’m using instance groups. I’ll change it in the future

If you invoke a Flow from the main Flow, you need to define the resource manager or executor at the subflow level. Man have you seen Orion though? Cuz Orion lets you define this from the main flow level. Is your thinking just very in line with Orion? 😆

I would approach it exactly the way you described with a separate flow run for each such use case. It's actually a very common and even recommended by AWS approach to have ephemeral EMR clusters per run/job.

@Anna Geller creating a cluster takes a lot of time, so reusing makes a lot of sense and it’s a common practise. One can do it in different ways; but even Databricks recenly added this ability: https://databricks.com/blog/2022/02/04/saving-time-and-costs-with-cluster-reuse-in-databricks-jobs.html

I personally haven’t seen it done. You are doing something like this right?

with Flow(...) as flow:
    with MyResource(...) as resource:
        resource.wait_emr_cluster_is_running()...
        some_task(resource)
        other_task(resource)

I don’t think it can be done because the Flow block should only contain tasks. MyResource there is a task so execution is deferred to the flow execution but using a method that is not a task will not defer

exactly! Maybe if I wrap the resource in another class; something like:

class myEMRUtils:
   
   @staticmethod
   def wait_emr_cluster_is_running(...):
     return result

   @resourcemanager
    class EMRCluster:
        def __init__(self,..)
        def setup(self, ...)
        def cleanup(self,...)

I dont think this works well, but not 100% sure. I would suggest calling

wait_emr_cluster_is_running

inside startup instead

that’s also a possible solution

Ah I see. You need to get the

add_step

and

wait_step

to function as tasks