dagster-rs/testing_notebook.py at main · mlops-itba/dagster-rs · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
# %%
import mlflow
logged_model = 'models:/keras_dot_product_model/7'

# Load model as a PyFuncModel.
loaded_model = mlflow.pyfunc.load_model(logged_model)
# %%

# %%
# Predict on a Pandas DataFrame.
import numpy as np
loaded_model.predict([np.array([1, 2]), np.array([2, 3])])
# %%
# %%
mlflow.set_experiment('test')
# %%
# %%
mlflow.start_run()
# %%
from tensorflow.keras import layers, Model
# %%
input = layers.Input(shape=(1,))
dense = layers.Dense(10)(input)
model = Model(input, dense)
# %%
model.summary()
# %%
logged_model = mlflow.tensorflow.log_model(
        model,
        "simple",
        registered_model_name='simple'
    )
# %%
logged_model.__dir__()
# %%

# %%
{
    'model_uri': logged_model.model_uri,
    'run_id': logged_model.run_id
}
# %%
logged_model.flavors
# %%
from movies_rs_dagster.assets.recommender.train_model import log_model
# %%
from movies_rs_dagster.assets.core.movies_users import movies, users, scores
from dagster import build_op_context
from movies_rs_dagster import job_configs
from dagster_mlflow import mlflow_tracking
# %%
from movies_rs_dagster import defs
from dagster import ExecuteInProcessResult
# %%
get_data_job = defs.get_job_def('get_data')
result = get_data_job.execute_in_process()
# %%
assert isinstance(result, ExecuteInProcessResult)
assert result.success
data = result.output_for_node('training_data')
# %%
result.asset_observations_for_node('preprocessed_training_data')
# %%
data
# %%
result.
# %%
y_pred = loaded_model.predict([
            data.user_id[:10],
            data.movie_id[:10]
        ])
# %%
((data.rating[:10].values - y_pred.reshape(-1))**2).sum()/10
# %%
from sklearn.metrics import mean_squared_error
# %%
mean_squared_error(y_pred.reshape(-1), data.rating[:10].values)
# %%
# %%
# %%

# %%
result.asset_value('user2id')
# %%
result = get_data_job
# %%
job_configs['resources']
# %%
context = build_op_context(
    op_config={'uri': 'https://raw.githubusercontent.com/mlops-itba/Datos-RS/main/data/peliculas_0.csv'}

)
# %%
users_out = users(context)
movies_out = movies(context)
# %%
scores(build_op_context(
    resources_config=job_configs['resources'],
    resources={'mlflow': mlflow_tracking}
))
# %%
# %%
users_out.value
# %%
# %%
job_configs['resources']

# %%