https://github.com/stitchfix/hamilton logo
Join Slack
Powered by
If the existence of `SOURCE1_HIST_FILE` determines...
# hamilton-help
s
If the existence of 
SOURCE1_HIST_FILE
determines the 
first_run
flag in the 
config
, and in the end, 
load_source1_data.py
will be a part of a 
subdag
. How can I make sure that every 
save_to
node saves out the file, each time it goes through that node?
Copy code
# load_source1_data.py

@config.when(first_run=True)
@save_to.parquet(path=value(SOURCE1_HIST_FILE), output_name_="saved_current_data_source1")
def current_data_source1__first_run(
    processed_data_source1: Collect[pd.DataFrame]
) -> pd.DataFrame:
    return pd.concat(processed_data_source1, axis="index", ignore_index=True)


@config.when_not(first_run=True)
@load_from.parquet(path=value(SOURCE1_HIST_FILE), inject_="last_run_source_source1")
@save_to.parquet(path=value(SOURCE1_HIST_FILE), output_name_="saved_current_data_source1")
def current_data_source1__subsq_run(
    processed_data_source1: Collect[pd.DataFrame], last_run_data_source1: pd.DataFrame
) -> pd.DataFrame:
    processed_df = pd.concat(processed_data_source1, axis="index", ignore_index=True)
    return pd.concat([last_run_source_source1, processed_df], axis="index", ignore_index=True)
s
IIUC you just need to request “saved_current_data_source1” as part of the subdag. That will return a 
dict
IIRC.
s
Is there an easy way to get all data saver nodes from the dag? maybe this is what i want:
Copy code
[v for v in dr.list_available_variables() if v.tags.get("hamilton.data_saver")]
I'd like to do this:
Copy code
savers = [node for node in dr.what_is_upstream_of("current_data_source1") if node.tags.get("hamilton.data_saver")]

dr.execute(savers)
However, 
saved_current_data_source1
cannot be seen. since i guess technically its not upstream.
s
Try something like this:
Copy code
[v for v in dr.list_available_variables() if v.tags.get("hamilton.data_saver") and "current_data_source1" in v.name]
the node is there, it’ll just be prefixed.
s
Copy code
@subdag(
    load_source1_data, load_source2_data
)
def norm_data(
    current_data_source1: pd.DataFrame,
    current_data_source2: pd.DataFrame
) -> pd.DataFrame:
    current_data_source1, current_data_source2
    # some more work
    return ...


final_node = "norm_data"
final_df = dr.execute([final_node])
savers = [v for v in dr.what_is_upstream_of(final_node) if v.tags.get("hamilton.data_saver") or final_node in v.name] 

dr.execute(savers)
So if i had multiple subdags, this could be a way to save every 
hamliton.data_saver
node it went though?
s
What's your goal? To ensure every saver is called?
s
yes sir
s
Then two options: 1. include it in the subdag function - this will mean it will always be run if you request that subdag to be run. 2. call it at execute time Subdag function:
Copy code
@subdag(
    load_source1_data, load_source2_data
)
def norm_data(
    current_data_source1: pd.DataFrame,
    current_data_source2: pd.DataFrame,
    saved_current_data_source1: dict # declare but don't use it...
) -> pd.DataFrame:
    current_data_source1, current_data_source2
    # some more work
    return ...
As you have with finding the tags on nodes and then requesting them.
Copy code
outputs = [v for v in dr.list_available_variables() if v.tags.get("hamilton.data_saver")]
dr.execute(outputs, inputs=...)
👍 1
s
Copy code
final_node = "norm_data"
final_df = dr.execute([final_node])
savers = [f"saved_{final_node}"] + [v for v in dr.what_is_upstream_of(final_node) if v.tags.get("hamilton.data_saver")]  

dr.execute(savers)
Should I expect this not to work? I think its getting hung up in the driver.
s
I think its getting hung up in the driver.
which line? Did you mean to do:
final_df = dr.execute([final_node])
that is executing things?
s
here 
dr.execute(savers)
s
did you print 
savers
? does it contain what you want?
s
yeahh its empty
Copy code
savers_all = {v.name for v in dr.list_available_variables() if v.tags.get("hamilton.data_saver")}
    savers = [v.name for v in dr.what_is_upstream_of(f"saved_{final_node}") if v.tags.get("hamilton.data_saver") and v.name in savers_all]
this was an idea, but it appears that 
dr.what_is_upstream_of
upstream node results that have 
save_to
attached to it does not reveal itself to the dag
s
Seth you are a 🐛 hunter extraordinaire ! Actually I think this is expected behavior
so yeah 
what_is_upstream_of
doesn’t return everything
since technically it isn’t upstream
if you look at the graph
{v.name for v in dr.list_available_variables() if v.tags.get(“hamilton.data_saver”)}
this should include the 
save_to
nodes
v.name for v in dr.what_is_upstream_of(f”saved_{final_node}“)
this wont — because the save_to nodes aren’t upstream of it.
norm_data.NAME_OF_SAVE_TO
should be what the node names look like
s
ok thats fair. and what you just sent is probably enough. but is there a way to get that behavior? that is, return all 
save_nodes
upstream of and attached to the node I am asking for?
s
well why 
upstream
? why not 
downstream
?
otherwise just use tags?
Copy code
@tag_outputs(foobar={"tag_key": "tag_value"})
@save_to.csv(path=source("fpath"), output_name_="foobar")
def a(input: pd.Series) -> pd.DataFrame:
    return input % 7
then search for that tag key & value.
s
upstream
because I am going to be asking for the "sink"/outermost node
will that tag execute the 
save_to
?
s
you’d just filter for it (https://hamilton.dagworks.io/en/latest/reference/drivers/Driver/#hamilton.driver.Driver.list_available_variables)
Copy code
[v for v in dr.list_available_variables(tag_filter={"tag_key": "tag_value", "hamilton.data_saver": None})]
s
ah ok, ty sir
OK, last question of the day.. Should I expect the body of this function to not execute because of 
pipe
? Or is 
pipe
not compatible with 
Parallelizable
?
Copy code
@config.when_not(first_run=True)
@tag(table="raw")
@pipe(
    step(
        _filter_from_highwatermark, date_col="COB_Date", highwatermark_date=source("highwatermark_date")
    )
)
@save_to.parquet(path=value(HIST_FILE_RAW), output_name_="saved_current_data_raw")
def raw_volumes_pre__subsq_run(
    file_path: str, read_fn: callable) -> pd.DataFrame:
    return read_fn(file_path).assign(COB_Date=lambda _df: pd.to_datetime(_df["COB_Date"]), SOURCE_FILE=file_path)
Copy code
ValueError: Error: raw_volumes_pre.with_filter_from_highwatermark is expecting file_path:<class 'pandas.core.frame.DataFrame'>, but found file_path:hamilton.htypes.Parallelizable[str].
Hamilton does not consider these types to be equivalent. If you believe they are equivalent, please reach out to the developers. Note that, if you have types that are equivalent for your purposes, you can create a graph adapter that checks the types against each other in a more lenient manner.
s
can you explain the code a bit more? 
file_path
is a string — but with 
_filter_from_highwatermark
it’s annotated as a dataframe?
s
The 
file_path
comes from upstream 
Parallelizable[str]
node. The 
body
of the function reads the data, and I want 
pipe
to filter the results via 
_filter_from_highwatermark
. I believe that’s everything that’s trying to occur here. Let me know if any of those expectations are wrong and if/how that can/could happen.
s
I’m trying to understand:
ValueError: Error: raw_volumes_pre.with_filter_from_highwatermark is expecting file_path:<class ‘*pandas.core.frame.DataFrame*’>, but found file_path:hamilton.htypes.Parallelizable[str].
but your code above doesn’t show that
s
Copy code
@config.when_not(first_run=True)
def file_path__first_run(files_to_load: Optional[list] = None) -> Parallelizable[str]:
    for f in files_to_load:
        yield f
is this what youre missing?
s
What is the 
_filter_from_highwatermark
function?
s
Copy code
def _filter_from_highwatermark(
    df: pd.DataFrame, date_col: str, highwatermark_date: datetime
) -> pd.DataFrame:
    """Filter the new data where `date_col` > `highwatermark_date` since each new file's data is MTD, if this is ran each day only the new data rows are needed."""
    return df.set_index(date_col).loc[lambda dt: dt > highwatermark_date].reset_index()
s
@Seth Stokes how were you thinking it’s supposed to work? 
_filter_from_highwatermark
expects a dataframe, not a string. e.g. you have file_path__first_run -> outputting a string -> going into _filter_from_highwatermark that expects a dataframe.
s
Does it not operate on the attached node 
raw_volumes_pre
?
s
pipe
runs before, i.e. it uses the first argument as the dependency to modify and return.
post_pipe
will do what you want (it’s currently under active development)
s
ok sweet. so for now, create an intermediate node
Open in Slack
PreviousNext
Powered by