Ok, I have a pretty complex POC that does what you’re looking at but I’m not convinced its the best approach. Some ideas ahead, realizing that this is something that we should improve in Hamilton — add the ability to make a post-hoc change to a node and have it managed through the framework. IMO this isn’t the best for pipelines — if you’re dropping rows you want it to be explicit, and not done after the fact. And data validation isn’t really meant to be a mutation operation (although it can be…). I’d suggest: 1. Centralizing/adding specific data validation/dropping nodes that do both 2. Then just letting it run and be smart about joining This could be as simple as:

def ratio(a: pd.Series, b: pd.Series) -> pd.Series:
    return a/b

def ratio_cleaned(ratio: pd.Series) -> pd.Series:
    return ratio[(ratio < 100) & (ratio > 0)]

Or, define your own decorator that mutates, you can leverage pandera:

@validate_and_drop(pandera_schema) # pandera allows mutation, this doesn't leverage hamilton to do this
def ratio(a: pd.Series, b: pd.Series) -> pd.Series:
    return a/b

In fact, in the case above, you could use something like

pipe

but its a little verbose and still requires a second one. Good if you have lots or want them to be Hamilton nodes.

def ratio_unprocessed(a: pd.Series, b: pd.Series) -> pd.Series:
    return a/b

@pipe(
    step(_with_outliers_removed, range=(0,100)),
)
def processed_ratio(ratio_unprocessed: pd.series) -> pd.Series:
    return raw_data

Finally here’s a POC of using data validators: 1. Writing a custom data validator that I want to use 2. Separating out the two types of nodes (created data validation nodes versus requested ones) 3. Using the pandas result builder to build the core result 4. Using the other results to filter (I don’t implement this but its easy) Code at bottom. Note you can use a custom ResultBuilder, but the tags aren’t exposed yet (they will be soon) so you have to be a bit clever.

from typing import Type

import pandas as pd

from hamilton import driver, base
from hamilton.base import PandasDataFrameResult
from hamilton.data_quality.base import DataValidator, ValidationResult
from hamilton.function_modifiers import check_output_custom


class FilteringRangeValidator(DataValidator):
    def __init__(self, range: tuple):
        super(FilteringRangeValidator, self).__init__(importance="warn")
        self.range = range

    def applies_to(self, datatype: Type[Type]) -> bool:
        return datatype == pd.Series

    def description(self) -> str:
        return "example data validator"

    @classmethod
    def name(cls) -> str:
        return "filtering_data_validator"

    def validate(self, dataset: pd.DataFrame) -> ValidationResult:
        return ValidationResult(
            passes=True,
            message="Data is valid, dropping some rows.",
            diagnostics={"rows_to_drop": dataset[(dataset < 0) | (dataset > 10)].index}  # drop the first 10 rows as an example
        )


@check_output_custom(FilteringRangeValidator(range=(0, 100)))
def ratio(a: pd.Series, b: pd.Series) -> pd.Series:
    """ratio"""
    return a / b


if __name__ == "__main__":
    import __main__

    dr = driver.Builder().with_modules(__main__).with_adapters(base.DictResult()).build()
    data_nodes = ["ratio"]
    validators = [item.name for item in dr.list_available_variables(tag_filter={"hamilton.data_quality.source_node": data_nodes})]
    raw_results = dr.execute(validators + data_nodes, inputs=pd.DataFrame({"a": [1.0, 2.0, 3.0], "b": [2.0, 0, 4.0]}))
    result_builder = PandasDataFrameResult()  # just reuse the result builder from hamilton
    results_to_merge = {k: v for k, v in raw_results.items() if k in data_nodes}
    merged_results = result_builder.build_result(**results_to_merge)
    rows_to_drop = {k: v.diagnostics["rows_to_drop"] for k, v in raw_results.items() if k in validators}
    for key, row in rows_to_drop.items():
        merged_results = merged_results.drop(row)
    print(merged_results)

Opened up https://github.com/DAGWorks-Inc/hamilton/issues/701 to track — feel free to comment if you think this would solve your problem (I think its much cleaner than what I presented above)