Node Parameters¶
Up to this point we've dealt with defining nodes and composing them together into pipelines. One issue is that all of the logic is static, the pipelines are unable to changed by external sources. Node Parameters are the solution to this- they comprise named arguments of a node that are defined at the point where the pipeline is run.
Definition¶
In [1]:
Copied!
import pandas as pd
from flypipe.node import node
@node(type="pandas")
def t1(param1):
print(f"param1: `{param1}`")
return pd.DataFrame()
@node(
type="pandas",
dependencies=[t1]
)
def t2(t1):
return t1
df = t2.run(parameters={t1: {"param1": "hello"}})
import pandas as pd
from flypipe.node import node
@node(type="pandas")
def t1(param1):
print(f"param1: `{param1}`")
return pd.DataFrame()
@node(
type="pandas",
dependencies=[t1]
)
def t2(t1):
return t1
df = t2.run(parameters={t1: {"param1": "hello"}})
param1: `hello`
Note just like any python function argument we can give a default value to use if no value is provided at runtime.
In [2]:
Copied!
import pandas as pd
from flypipe.node import node
@node(type="pandas")
def t1(param1=None):
print(f"param1: `{param1}`")
return pd.DataFrame()
df = t1()
import pandas as pd
from flypipe.node import node
@node(type="pandas")
def t1(param1=None):
print(f"param1: `{param1}`")
return pd.DataFrame()
df = t1()
param1: `None`
Parameters and node functions¶
Parameters can also be used with node functions, together you have more flexibility on dynamically changing not only node behaviours but also graph compositions
In [3]:
Copied!
from flypipe import node_function
@node_function()
def t1_func(node_selector="t1"):
@node(type="pandas")
def t1():
return pd.DataFrame()
if node_selector == "t1":
return t1
@node(
type="pandas",
dependencies=[t1]
)
def t2():
return pd.DataFrame()
return t2, t1
displayHTML(t1_func.html())
from flypipe import node_function
@node_function()
def t1_func(node_selector="t1"):
@node(type="pandas")
def t1():
return pd.DataFrame()
if node_selector == "t1":
return t1
@node(
type="pandas",
dependencies=[t1]
)
def t2():
return pd.DataFrame()
return t2, t1
displayHTML(t1_func.html())
Out[3]:
Changing parameter node_selector¶
In [4]:
Copied!
displayHTML(
t1_func.html(
parameters={
t1_func: {'node_selector': 't2'}
}
))
displayHTML(
t1_func.html(
parameters={
t1_func: {'node_selector': 't2'}
}
))
Out[4]:
Suggested use cases for node parameters¶
- you need the node to behave differently accordingly to some settings
- train a model inside a node with changeable splits of data between train, test and validation
- run nodes that execute customisable versions of an mlflow model by encoding the model run id into a node parameter
train a model with different splits of data (train, test and validation)
In [5]:
Copied!
from flypipe.node import node
from flypipe.schema import Schema, Column
from flypipe.schema.types import Date
import pandas as pd
from datetime import datetime, timedelta
@node(
type="pandas",
output=Schema([
Column("date", Date(), "date"),
])
)
def my_model_features():
date = datetime(2022,1,1)
dates = [date + timedelta(days=day) for day in range(11)]
return pd.DataFrame(data={"date": dates})
@node(
type="pandas",
dependencies=[my_model_features]
)
def tag_data(my_model_features, split={'train': datetime(2022,1,4), "test": datetime(2022,1,7)}):
print(f"""
Split data
----------
train: date <= {split['train'].date()}
test: {split['train'].date()} < date <= {split['test'].date()}
val: {split['test'].date()} < date
""")
my_model_features["tag"] = "val"
my_model_features.loc[my_model_features['date'] <= split['test'], "tag"] = "test"
my_model_features.loc[my_model_features['date'] <= split['train'], "tag"] = "train"
return my_model_features
@node(
type="pandas",
dependencies=[tag_data]
)
def my_model(tag_data):
return tag_data
df = my_model.run()
display(df)
from flypipe.node import node
from flypipe.schema import Schema, Column
from flypipe.schema.types import Date
import pandas as pd
from datetime import datetime, timedelta
@node(
type="pandas",
output=Schema([
Column("date", Date(), "date"),
])
)
def my_model_features():
date = datetime(2022,1,1)
dates = [date + timedelta(days=day) for day in range(11)]
return pd.DataFrame(data={"date": dates})
@node(
type="pandas",
dependencies=[my_model_features]
)
def tag_data(my_model_features, split={'train': datetime(2022,1,4), "test": datetime(2022,1,7)}):
print(f"""
Split data
----------
train: date <= {split['train'].date()}
test: {split['train'].date()} < date <= {split['test'].date()}
val: {split['test'].date()} < date
""")
my_model_features["tag"] = "val"
my_model_features.loc[my_model_features['date'] <= split['test'], "tag"] = "test"
my_model_features.loc[my_model_features['date'] <= split['train'], "tag"] = "train"
return my_model_features
@node(
type="pandas",
dependencies=[tag_data]
)
def my_model(tag_data):
return tag_data
df = my_model.run()
display(df)
Split data
----------
train: date <= 2022-01-04
test: 2022-01-04 < date <= 2022-01-07
val: 2022-01-07 < date
| date | tag | |
|---|---|---|
| 0 | 2022-01-01 | train |
| 1 | 2022-01-02 | train |
| 2 | 2022-01-03 | train |
| 3 | 2022-01-04 | train |
| 4 | 2022-01-05 | test |
| 5 | 2022-01-06 | test |
| 6 | 2022-01-07 | test |
| 7 | 2022-01-08 | val |
| 8 | 2022-01-09 | val |
| 9 | 2022-01-10 | val |
| 10 | 2022-01-11 | val |
running with different splits
In [6]:
Copied!
df = my_model.run(parameters={
tag_data: {
'split': {'train': datetime(2022,1,2), "test": datetime(2022,1,4)}
}
})
display(df)
df = my_model.run(parameters={
tag_data: {
'split': {'train': datetime(2022,1,2), "test": datetime(2022,1,4)}
}
})
display(df)
Split data
----------
train: date <= 2022-01-02
test: 2022-01-02 < date <= 2022-01-04
val: 2022-01-04 < date
| date | tag | |
|---|---|---|
| 0 | 2022-01-01 | train |
| 1 | 2022-01-02 | train |
| 2 | 2022-01-03 | test |
| 3 | 2022-01-04 | test |
| 4 | 2022-01-05 | val |
| 5 | 2022-01-06 | val |
| 6 | 2022-01-07 | val |
| 7 | 2022-01-08 | val |
| 8 | 2022-01-09 | val |
| 9 | 2022-01-10 | val |
| 10 | 2022-01-11 | val |