forked from mlflow/mlflow
-
Notifications
You must be signed in to change notification settings - Fork 1
/
run_checks.py
67 lines (51 loc) · 1.87 KB
/
run_checks.py
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
import os
import argparse
import sqlalchemy
from sqlalchemy.schema import MetaData, CreateTable
import mlflow
from mlflow.tracking._tracking_service.utils import _TRACKING_URI_ENV_VAR
class MockModel(mlflow.pyfunc.PythonModel):
def load_context(self, context):
pass
def predict(self, context, model_input):
pass
def parse_args():
parser = argparse.ArgumentParser()
parser.add_argument("--schema-output", required=True, help="Output path of DB schema")
return parser.parse_args()
def run_logging_operations():
with mlflow.start_run() as run:
print("Tracking URI:", mlflow.get_tracking_uri())
mlflow.log_param("p", "param")
mlflow.log_metric("m", 1.0)
mlflow.set_tag("t", "tag")
mlflow.pyfunc.log_model(
artifact_path="model",
python_model=MockModel(),
registered_model_name="mock",
)
print(mlflow.get_run(run.info.run_id))
def get_db_schema():
engine = sqlalchemy.create_engine(mlflow.get_tracking_uri())
created_tables_metadata = MetaData(bind=engine)
created_tables_metadata.reflect()
# Write out table schema as described in
# https://docs.sqlalchemy.org/en/13/faq/metadata_schema.html#how-can-i-get-the-create-table-drop-table-output-as-a-string
lines = []
for ti in created_tables_metadata.sorted_tables:
lines += list(map(str.rstrip, str(CreateTable(ti)).splitlines()))
return "\n".join(lines)
def main():
assert _TRACKING_URI_ENV_VAR in os.environ
args = parse_args()
run_logging_operations()
schema = get_db_schema()
title = "Schema"
print("=" * 10, title, "=" * 10)
print(schema)
print("=" * (20 + 2 + len(title)))
os.makedirs(os.path.dirname(args.schema_output), exist_ok=True)
with open(args.schema_output, "w") as f:
f.write(schema)
if __name__ == "__main__":
main()