Correct Hamilton pattern when a data pipeline requires merging two DataFrames?

[filpia]

Hi!

I'm just getting started with Hamilton and the existing documentation is great for illustrating the actions performed on a single df. If in my data pipeline I wanted to initialize a df, perform some feature engineering on it, merge a separate df onto this (potentially using engineered features as join keys), and then continuing to engineer features on this newly joined df, what's the proper pattern here? Is there syntax in hamilton to natively incorporate merge functionality into its DAGs or no?

My gut instinct would be to

1. use separate drivers to take separate dataframes as far as I need to
2. merge the resultant dataframes using native pd.merge() or df1.merge(df2, on=...)
3. initialize a new driver to engineer features on the dataframe output by the previous step

[skrawcz]

@filpia so sorry for the very very slow response; let me fix my notification settings so it doesn't happen again.

Good question, and there's a few approaches here; at a high level, with Hamilton you could model this process entirely through functions. The different approaches therefore are really just determining the boundaries of the DAG you're grouping/curating:

• Your solution, as you mention, seems like a perfectly valid approach. If the "glue" code of merging is context specific business logic, then that's not a bad way to keep those parts quite separate.

  dr1 = driver.Driver({...}, feature_module1, ...)
  df1 = dr1.execute(...)
  dr2 = driver.Driver({...}, feature_module2, ...)
  df2 = dr2.execute(...)
  merged_df = pd.merge(df1, df2, ...)
  dr3 = driver.Driver({col: merged_df[col] for col  in merged_df.columns}, other_features, ...)
  df3 = dr3.execute(...)

• You could encode step (2) in your approach as a named Hamilton function itself, skip doing three drivers, and have just one. This has the consequence of requiring an explicit update if new features are added, and figuring out how to name features before and after the function. But perhaps this extra friction ensures clarity when changes come through. Note: you could still do the above approach with three drivers, you'd just need to either use the overrides= functionality, and/or curate the functions distinct into modules.

   @extract_columns(...)
   def merged_df(col1: ..., col2: ..., colN) -> pd.DataFrame:
    # df = ... logic to merge
    return df

• You could build a custom results builder that takes care of the merge, but this requires using two drivers and would roughly look like the following. You can think of the result builders as a custom "reduce" step at the end of execution. By default Hamilton creates a Pandas dataframe, but you can augment that logic yourself...

  dr1 = driver.Driver({...}, feature_module1, feature_module2, adapter=base.SimplePythonGraphAdapter(my_custom_results_builder(feature_set1, feature_set2)))
  merged_df = dr1.execute(feature_set1 + feature_set2)
  dr3 = driver.Driver({col: merged_df[col] for col  in merged_df.columns}, other_features, ...)
  df3 = dr3.execute(...)

• The other approaches would be some combination of the above, and it depends on where you want to draw the line with function curation. The different approaches do impact how code would be updated/changed in your system, so that would be the thing to feel out. E.g. we have this issue (Tag-based dependencies #226) open that could make a named function easier to manage in this scenario -- please provide us feedback if you have something to add.