h_ray.RayGraphAdapter

The graph adapter to delegate execution of the individual nodes in a Apache Hamilton graph to Ray.

class hamilton.plugins.h_ray.RayGraphAdapter(result_builder: ResultMixin, ray_init_config: dict[str, Any] = None, shutdown_ray_on_completion: bool = False)[source]

Class representing what’s required to make Hamilton run on Ray.

This walks the graph and translates it to run onto Ray.

Use pip install apache-hamilton[ray] to get the dependencies required to run this.

Use this if:

• you want to utilize multiple cores on a single machine, or you want to scale to larger data set sizes with a Ray cluster that you can connect to. Note (1): you are still constrained by machine memory size with Ray; you can’t just scale to any dataset size. Note (2): serialization costs can outweigh the benefits of parallelism so you should benchmark your code to see if it’s worth it.

Notes on scaling:

• Multi-core on single machine ✅
• Distributed computation on a Ray cluster ✅
• Scales to any size of data ⛔️; you are LIMITED by the memory on the instance/computer 💻.

Function return object types supported:

• Works for any python object that can be serialized by the Ray framework. ✅

Pandas?

• ⛔️ Ray DOES NOT do anything special about Pandas.

CAVEATS

• Serialization costs can outweigh the benefits of parallelism, so you should benchmark your code to see if it’s worth it.

DISCLAIMER – this class is experimental, so signature changes are a possibility!

__init__(result_builder: ResultMixin, ray_init_config: dict[str, Any] = None, shutdown_ray_on_completion: bool = False)[source]

Constructor

You have the ability to pass in a ResultMixin object to the constructor to control the return type that gets produce by running on Ray.

Parameters:

• result_builder – Required. An implementation of base.ResultMixin.
• ray_init_config – allows to connect to an existing cluster or start a new one with custom configuration (https://docs.ray.io/en/latest/ray-core/api/doc/ray.init.html)
• shutdown_ray_on_completion – by default we leave the cluster open, but we can also shut it down

do_build_result(outputs: dict[str, Any]) → Any[source]

Builds the result and brings it back to this running process.

Parameters:

outputs – the dictionary of key -> Union[ray object reference | value]

Returns:

The type of object returned by self.result_builder.

static do_check_edge_types_match(type_from: type, type_to: type) → bool[source]

Method that checks whether two types are equivalent. This is used when the function graph is being created.

Parameters:

• type_from – The type of the node that is the source of the edge.
• type_to – The type of the node that is the destination of the edge.

Return bool:

Whether or not they are equivalent

do_remote_execute(*, execute_lifecycle_for_node: Callable, node: Node, **kwargs: dict[str, Any]) → Any[source]

Function that is called as we walk the graph to determine how to execute a hamilton function.

Parameters:

• execute_lifecycle_for_node – wrapper function that executes lifecycle hooks and methods
• kwargs – the arguments that should be passed to it.

Returns:

returns a ray object reference.

static do_validate_input(node_type: type, input_value: Any) → bool[source]

Method that an input value maches an expected type.

Parameters:

• node_type – The type of the node.
• input_value – The value that we want to validate.

Returns:

Whether or not the input value matches the expected type.

post_graph_execute(*args, **kwargs)[source]

We have the option to close the cluster down after execution.