Hey team! I am also new to Hamilton, starting effort on a POC for Ray+Hamilton for our feature creation pipeline. One small question: I want to output a df using

PandasDataFrameResult

. The error I get is

ValueError: array length 10000 does not match index length 9871

. the code looks simple:

output_columns = [
        "ids",
        "sum_contributions",
    ]

    df = dr.execute(output_columns)

This makes sense because in

sum_contributions

I am filtering out some records. Is there a way to rewrite this to tell the Driver to do a left join - or another way to avoid this error? Is the only solution creating a function that handles all the joining logic and simply outputs the final df?

Yes - you can control how the final result is aggregated together via a custom result builder

I see. I was looking at that page and this made me check in first to see if there's a built-in method I'm missing

If you have a use case for a custom ResultBuilder, tell us on Slack or via a GitHub issues.

Yeah I realize I need to update the page a little to be clearer — here’s a sketch:

from hamilton import driver, base
import functions

import typing
from hamilton import lifecycle
class MyCustomBuilder(lifecycle.ResultBuilder):
     # add a constructor if you need to
     @staticmethod
     def build_result(**outputs: typing.Dict[str, typing.Any]) -> pd.DataFrame:
         """Custom function you fill in"""
         # your logic would go here
         return pd.DataFrame(outputs).head(1)

# Create the driver -- passing in an empty config and the functions module
dr = (
  driver.Builder()
    .with_config({})
    .with_modules(functions)
    .with_adapters(MyCustomBuilder())
    .build()
)

result = dr.execute(
  [
    "spend", 
    "signups", 
    "avg_3wk_spend"],
  inputs=load_data().to_dict(orient="series")
)

does this have any benefits over defining a node to do the joining logic? I assume its reusability and code organization -- any other performance benefits?

Yeah so there’s a duality between the result builder and a function/node in the DAG.

Got it, right so it'd be harder to abstract that portion. Makes sense. Thanks a ton for the prompt and helpful answer!

E.g. you could have a function that defines the feature set you’re creating — which could have some nice properties that then allow you to connect it downstream for model fitting for example

working on huge datasets (e.g. one table is ~274.52 GB / 87M rows). We want to generate 100-200 features based on this data (at least for now, I think we'd want to be able to support 1000s of features and 10s of different data sources). So we really need scalability for our feature calculations.

If you’re doing a bake off - consider trying polars on a single machine as a base line. > So we really need scalability for our feature calculations. Yeah the thing to think about is the datatransfer between processes has to be worth it. > another question around Ray while on this topic. Does ray simply provide multi-processing where each node works on a single function, or does it also do a single function’s calculations on many nodes? Can you disambiguate node & function for me here? Do you mean Hamilton node? or a machine that Ray is running on? does function refer to a Hamilton function? Just want to be sure since I use them interchangeably myself…

Does Ray solely support Hamilton task/function parallelism, where each Ray node executes distinct functions, or does it also enable data parallelism, allowing a single Hamilton function's computations to be distributed across multiple Ray nodes?

So we have both modes if I’m understanding correctly. The RayGraphAdapter (right now) takes a hamilton function and maps it to a remote task 1-1. The other is via Parallel & Collect. You can run a “sub-dag” as a Ray task, where you can send over partitions of data for that “sub-dag” to work on.

I'm going to look more closely as Parallel & collect. 1. does P&C allow dfs? 2. Can you elaborate on this: I don't fully understand it

If you're doing a bake off - consider trying polars on a single machine as a base line.

3.

Yeah the thing to think about is the datatransfer between processes has to be worth it.

What are the things to consider here? I know that we need some layer of distributed processing for the scale of data we use, but I don't fully understand this point. Is it ensuring the tasks are "beefy"? a. I know that Hamilton materializes the data in every node (hamilton function). This means the full data needs to fit in memory. Is this going to be a big blocker for us as we scale?

> does P&C allow dfs?

def df_batch(large_df: pd.DataFrame) -> Parallelizable[pd.DataFrame]:
    for i in range(0, len(df), BATCH_SIZE):
        yield df.iloc[i:i + BATCH_SIZE]

def my_func(df_batch: pd.DataFrame) -> ...

that should work > consider trying polars on a single machine as a base line. Polars is more memory efficient than Pandas. They have a way to “batch” over data with LazyFrame. It could give you a baseline to test a “distributed” solution against. > Is it ensuring the tasks are “beefy”? Yep. You need the task computation to be long enough so that the time that it takes to serialize and deserialize the data is worth parallelizing it. E.g. time to serialize and deserialize * N, should be less than the time to run the task in serial, where N is how many things in parallel you’re doing. > I know that Hamilton materializes the data in every node (hamilton function). This means the full data needs to fit in memory. Is this going to be a big blocker for us as we scale? No. You’d just change the data type / chunk the data appropriately. Hamilton is data type agnostic, it doesn’t care whether you’re using pandas, polars, pyspark, etc. So you’d use the data type / computation appropriate for your scale. Does that make sense? Feel free to come to office hours on Tuesday next week (9:30 AM PT) and someone can explain more there.

Got it, thank a ton!! Going to think through your answers more and definitely come to office hours to discuss. Are they every day?

Ah, they are on Tuesdays - unless we have a meet up.

cool thanks