https://github.com/stitchfix/hamilton logo
Join Slack
Powered by
Does the input and output type of a `pipe` `step` ...
# hamilton-help
s
Does the input and output type of a 
pipe
step
have to be the same? I am trying to do the below and it is failing on type checking.
Copy code
@pipe_output(
    step(_filter_bom_run).when(last_run="bom"),
    step(_filter_last_run).when(last_run="last"),
)
def child_process_run_id(child_process_runs: pd.DataFrame) -> int:
    """Return run id for process from `last_run` config."""
    return child_process_runs
The fn sig is the same for each step, 
DataFrame -> int
s
Copy code
def child_process_run_id(child_process_runs: pd.DataFrame) -> int:
should be:
Copy code
def child_process_run_id(child_process_runs: pd.DataFrame) -> pd.DataFrame:
I think
e
Yep Stefan is right, it is confusing though…
s
OK thanks, yeah I had to do switch to 
pipe_input
.
Copy code
@pipe_input(
    step(_filter_bom_run).when(last_run="bom"),
    step(_filter_last_run).when(last_run="last"),
)
def child_process_run_id(child_process_runs: pd.DataFrame) -> int:
    """Return run id for process from `last_run` logic."""
    return child_process_runs["id"].iloc[0]
j
Hi Seth, the following works when I run it
Copy code
from typing import Any, List
import pandas as pd
from hamilton.function_modifiers import pipe_output, step, hamilton_exclude

# data1 and data2
@hamilton_exclude
def filter_(some_data:pd.DataFrame)->pd.DataFrame:
    return some_data.dropna()

# data1 and data2
@hamilton_exclude
def reset_to_int_(some_data:pd.DataFrame)->int:
    return len(some_data)

@pipe_output(
    step(filter_),
    step(reset_to_int_),
)
def data_1()->pd.DataFrame:
    df = pd.DataFrame.from_dict({'col_1': [3, 2, <http://pd.NA|pd.NA>, 0], 'col_2': ['a', 'b', <http://pd.NA|pd.NA>, 'd']})
    return df

def feat_A(data_1:int)->int:
    return data_1


if __name__ == "__main__":
    from hamilton import driver
    import __main__ as user_module

    node_to_execute = ["data_1.raw", "feat_A"]

    dr = driver.Builder().with_modules(user_module).build()
    dr.visualize_execution(final_vars=node_to_execute, output_file_path='./DAG.png')
    results = dr.execute(final_vars=node_to_execute)

    for k,v in results.items():
        print(k,v)
And I get a dataframe for the 
.raw
node and an integer as the output:
Copy code
data_1.raw   col_1 col_2
0     3     a
1     2     b
2  <NA>  <NA>
3     0     d
feat_A 3
With respect to the type as @Stefan Krawczyk already pointed out for the initial node you need to chain them through correct, for your example the signature seem a bit off since you set int at every step; the following should work
Copy code
child_process_run_id(child_process_runs: pd.DataFrame) -> pd.DataFrame
_filter_bom_run(no_need_for_same_name: pd.DataFrame) --> pd.DataFrame
_filter_last_run(but_type_needs_to_mathc_previous_output:pd.DataFrame) --> int
Of course you can also set something different, just like in a normal hamilton dag
Copy code
child_process_run_id(child_process_runs: pd.DataFrame) -> str
_filter_bom_run(no_need_for_same_name: str) --> int
_filter_last_run(but_type_needs_to_mathc_previous_output:int) --> CustomClassType
s
Ok awesome, I’ll try this again tomorrow
2 Views
Open in Slack
PreviousNext
Powered by