https://github.com/stitchfix/hamilton logo
Join Slack
Powered by
I have a process like this but am not seeing the `...
# hamilton-help
s
I have a process like this but am not seeing the 
filtered_df.with_rename_columns
Copy code
@pipe_input(
    step(_rename_columns, mapping=NORMALIZED_MAPPING),
    step(_filter_out_noise, diff_col="Amount Diff").when_in(process_id=["a", "b", "c", "d"]), 
)
def filtered_df(raw: pd.DataFrame) -> pd.DataFrame:
    return raw
Am I missing something?
s
are you talking about the visualization? or?
s
I am not seeing it in the dag
s
This works for me:
Copy code
%%cell_to_module pipe_example --display

from hamilton.function_modifiers import pipe_input, step
import pandas as pd
NORMALIZED_MAPPING = {"old_name": "new_name"}

def _rename_columns(df: pd.DataFrame, mapping: dict) -> pd.DataFrame:
    return df.rename(columns={"old_name": "new_name"})

def _filter_out_noise(df: pd.DataFrame, diff_col: str) -> pd.DataFrame:
    return df[df[diff_col] > 0]

@pipe_input(
    step(_rename_columns, mapping=NORMALIZED_MAPPING),
    step(_filter_out_noise, diff_col="Amount Diff").when_in(process_id=["a", "b", "c", "d"]), 
)
def filtered_df(raw: pd.DataFrame) -> pd.DataFrame:
    return raw
s
Thanks for checking. I must have fatfingered something here because 
filtered_df
is now apparently absent from my dag.
👍 1
Copy code
# transforms 
@pipe_input(
    step(_rename_columns, mapping=NORMALIZED_MAPPING),
    step(_filter_out_noise, diff_col="Amount Diff").when_in(process_id=["a", "b", "c", "d"]), 
)
def filtered_df(raw: pd.DataFrame) -> pd.DataFrame:
    return raw

@hamilton_exclude
def all_processes_table(process_ids: List[Literal["a","b","c","d"]]) -> pd.DataFrame:
    import __main__
    from hamilton import driver, base 
    builder = (
        driver.Builder()
        .with_modules(__main__)
        .with_adapters(base.PandasDataFrameResult())
    )
    
    ALL_DFS= []
    for _id in process_ids:
        builder_copy = builder.copy()
        dr = builder_copy.with_config({"process_id": _id}).build()
        df = dr.execute(["filtered_df"], inputs={"save_location": save_location})
        ALL_DFS.append(df)
    return pd.concat(ADJUSTMENT_DFS, ignore_index=True)
The problem was 
__main__
when trying to use 
all_processes_table
in 
more_dataflow
Copy code
# more_dataflow.py
from transforms import all_processes_table

all_processes_table(process_ids=list("abcd"))
would only 
_rename_columns
step
be applied if the 
when_in
condition failed?
s
would only 
_rename_columns
step
be applied if the 
when_in
condition failed?
no, it’s a separate independent step.
s
ok I meant to say, would only 
_rename_columns
step
result return 
raw
in the even the sonditional subsequent step fails to be met?
1. 
raw -> _rename_columns -> filtered_df
- condtion not met for 
_filter_out_noise
2. 
raw -> _rename_columns -> _filter_out_noise -> filtered_df
- condition met for 
_filter_out_noise
s
correct
s
ok ty sir. any advice on the 
__main__
issue ? Thats a good POC workflow, but a bad practice for what I was trying to do?
s
If you import 
all_processes_table
from another module, 
__main__
will be different than the one it is in. E.g.
Copy code
print(__main__)
in that function.
👍 1
2 Views
Open in Slack
PreviousNext
Powered by