Skip to content

Commit 53ebe47

Browse files
oavdeevoavdeev
authored
Python-centric feast deploy CLI (#1362)
* Python-centric feast deploy CLI Signed-off-by: Oleg Avdeev <oleg.v.avdeev@gmail.com> * sqlite provider Signed-off-by: Oleg Avdeev <oleg.v.avdeev@gmail.com> * add a proper test for local Signed-off-by: Oleg Avdeev <oleg.v.avdeev@gmail.com> * gcp test Signed-off-by: Oleg Avdeev <oleg.v.avdeev@gmail.com> * add missing files and mark integration test Signed-off-by: Oleg Avdeev <oleg.v.avdeev@gmail.com> * reconcile configs Signed-off-by: Oleg Avdeev <oleg.v.avdeev@gmail.com> * add missing file Signed-off-by: Oleg Avdeev <oleg.v.avdeev@gmail.com> * add missing dep Signed-off-by: Oleg Avdeev <oleg.v.avdeev@gmail.com> * add missing dep Signed-off-by: Oleg Avdeev <oleg.v.avdeev@gmail.com> * fix deps again Signed-off-by: Oleg Avdeev <oleg.v.avdeev@gmail.com> * use datastore not firestore Signed-off-by: Oleg Avdeev <oleg.v.avdeev@gmail.com> * fix tests Signed-off-by: Oleg Avdeev <oleg.v.avdeev@gmail.com> * comments Signed-off-by: Oleg Avdeev <oleg.v.avdeev@gmail.com>
1 parent 321894d commit 53ebe47

File tree

14 files changed

+480
-58
lines changed

14 files changed

+480
-58
lines changed

‎sdk/python/feast/cli.py‎

Lines changed: 36 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,7 @@
1515
import json
1616
import logging
1717
import sys
18+
from pathlib import Path
1819
from typing import Dict
1920

2021
import click
@@ -26,6 +27,8 @@
2627
from feast.entity import Entity
2728
from feast.feature_table import FeatureTable
2829
from feast.loaders.yaml import yaml_loader
30+
from feast.repo_config import load_repo_config
31+
from feast.repo_operations import apply_total, registry_dump, teardown
2932

3033
_logger = logging.getLogger(__name__)
3134

@@ -353,5 +356,38 @@ def project_list():
353356
print(tabulate(table, headers=["NAME"], tablefmt="plain"))
354357

355358

359+
@cli.command("apply")
360+
@click.argument("repo_path", type=click.Path(dir_okay=True, exists=True))
361+
def apply_total_command(repo_path: str):
362+
"""
363+
Applies a feature repo
364+
"""
365+
repo_config = load_repo_config(Path(repo_path))
366+
367+
apply_total(repo_config, Path(repo_path).resolve())
368+
369+
370+
@cli.command("teardown")
371+
@click.argument("repo_path", type=click.Path(dir_okay=True, exists=True))
372+
def teardown_command(repo_path: str):
373+
"""
374+
Tear down infra for a feature repo
375+
"""
376+
repo_config = load_repo_config(Path(repo_path))
377+
378+
teardown(repo_config, Path(repo_path).resolve())
379+
380+
381+
@cli.command("registry-dump")
382+
@click.argument("repo_path", type=click.Path(dir_okay=True, exists=True))
383+
def registry_dump_command(repo_path: str):
384+
"""
385+
Prints contents of the metadata registry
386+
"""
387+
repo_config = load_repo_config(Path(repo_path))
388+
389+
registry_dump(repo_config)
390+
391+
356392
if __name__ == "__main__":
357393
cli()

‎sdk/python/feast/feature_store.py‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -13,7 +13,7 @@
1313
# limitations under the License.
1414
from typing import Optional
1515

16-
from feast.feature_store_config import Config
16+
from feast.repo_config import RepoConfig, load_repo_config
1717

1818

1919
class FeatureStore:
@@ -22,13 +22,13 @@ class FeatureStore:
2222
"""
2323

2424
def __init__(
25-
self, config_path: Optional[str], config: Optional[Config],
25+
self, config_path: Optional[str], config: Optional[RepoConfig],
2626
):
2727
if config_path is None or config is None:
2828
raise Exception("You cannot specify both config_path and config")
2929
if config is not None:
3030
self.config = config
3131
elif config_path is not None:
32-
self.config = Config.from_path(config_path)
32+
self.config = load_repo_config(config_path)
3333
else:
34-
self.config = Config()
34+
self.config = RepoConfig()

‎sdk/python/feast/feature_store_config.py‎

Lines changed: 0 additions & 53 deletions
This file was deleted.

‎sdk/python/feast/infra/__init__.py‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
1+
# from .provider import Provider

‎sdk/python/feast/infra/gcp.py‎

Lines changed: 76 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,76 @@
1+
from datetime import datetime
2+
from typing import List, Optional
3+
4+
from feast import FeatureTable
5+
from feast.infra.provider import Provider
6+
from feast.repo_config import DatastoreOnlineStoreConfig
7+
8+
9+
def _delete_all_values(client, key) -> None:
10+
"""
11+
Delete all data under the key path in datastore.
12+
"""
13+
while True:
14+
query = client.query(kind="Value", ancestor=key)
15+
entities = list(query.fetch(limit=1000))
16+
if not entities:
17+
return
18+
19+
for entity in entities:
20+
print("Deleting: {}".format(entity))
21+
client.delete(entity.key)
22+
23+
24+
class Gcp(Provider):
25+
_project_id: Optional[str]
26+
27+
def __init__(self, config: Optional[DatastoreOnlineStoreConfig]):
28+
if config:
29+
self._project_id = config.project_id
30+
else:
31+
self._project_id = None
32+
33+
def _initialize_client(self):
34+
from google.cloud import datastore
35+
36+
if self._project_id is not None:
37+
return datastore.Client(self.project_id)
38+
else:
39+
return datastore.Client()
40+
41+
def update_infra(
42+
self,
43+
project: str,
44+
tables_to_delete: List[FeatureTable],
45+
tables_to_keep: List[FeatureTable],
46+
):
47+
from google.cloud import datastore
48+
49+
client = self._initialize_client()
50+
51+
for table in tables_to_keep:
52+
key = client.key("FeastProject", project, "FeatureTable", table.name)
53+
entity = datastore.Entity(key=key)
54+
entity.update({"created_at": datetime.utcnow()})
55+
client.put(entity)
56+
57+
for table in tables_to_delete:
58+
_delete_all_values(
59+
client, client.key("FeastProject", project, "FeatureTable", table.name)
60+
)
61+
62+
# Delete the table metadata datastore entity
63+
key = client.key("FeastProject", project, "FeatureTable", table.name)
64+
client.delete(key)
65+
66+
def teardown_infra(self, project: str, tables: List[FeatureTable]) -> None:
67+
client = self._initialize_client()
68+
69+
for table in tables:
70+
_delete_all_values(
71+
client, client.key("FeastProject", project, "FeatureTable", table.name)
72+
)
73+
74+
# Delete the table metadata datastore entity
75+
key = client.key("FeastProject", project, "FeatureTable", table.name)
76+
client.delete(key)

‎sdk/python/feast/infra/local_sqlite.py‎

Lines changed: 36 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,36 @@
1+
import os
2+
import sqlite3
3+
from typing import List
4+
5+
from feast import FeatureTable
6+
from feast.infra.provider import Provider
7+
from feast.repo_config import LocalOnlineStoreConfig
8+
9+
10+
def _table_id(project: str, table: FeatureTable) -> str:
11+
return f"{project}_{table.name}"
12+
13+
14+
class LocalSqlite(Provider):
15+
_db_path: str
16+
17+
def __init__(self, config: LocalOnlineStoreConfig):
18+
self._db_path = config.path
19+
20+
def update_infra(
21+
self,
22+
project: str,
23+
tables_to_delete: List[FeatureTable],
24+
tables_to_keep: List[FeatureTable],
25+
):
26+
conn = sqlite3.connect(self._db_path)
27+
for table in tables_to_keep:
28+
conn.execute(
29+
f"CREATE TABLE IF NOT EXISTS {_table_id(project, table)} (key BLOB, value BLOB)"
30+
)
31+
32+
for table in tables_to_delete:
33+
conn.execute(f"DROP TABLE IF EXISTS {_table_id(project, table)}")
34+
35+
def teardown_infra(self, project: str, tables: List[FeatureTable]) -> None:
36+
os.unlink(self._db_path)

‎sdk/python/feast/infra/provider.py‎

Lines changed: 49 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,49 @@
1+
import abc
2+
from typing import List
3+
4+
from feast import FeatureTable
5+
from feast.repo_config import RepoConfig
6+
7+
8+
class Provider(abc.ABC):
9+
@abc.abstractmethod
10+
def update_infra(
11+
self,
12+
project: str,
13+
tables_to_delete: List[FeatureTable],
14+
tables_to_keep: List[FeatureTable],
15+
):
16+
"""
17+
Reconcile cloud resources with the objects declared in the feature repo.
18+
19+
Args:
20+
tables_to_delete: Tables that were deleted from the feature repo, so provider needs to
21+
clean up the corresponding cloud resources.
22+
tables_to_keep: Tables that are still in the feature repo. Depending on implementation,
23+
provider may or may not need to update the corresponding resources.
24+
"""
25+
...
26+
27+
@abc.abstractmethod
28+
def teardown_infra(self, project: str, tables: List[FeatureTable]):
29+
"""
30+
Tear down all cloud resources for a repo.
31+
32+
Args:
33+
tables: Tables that are declared in the feature repo.
34+
"""
35+
...
36+
37+
38+
def get_provider(config: RepoConfig) -> Provider:
39+
if config.provider == "gcp":
40+
from feast.infra.gcp import Gcp
41+
42+
return Gcp(config.online_store.datastore)
43+
elif config.provider == "local":
44+
from feast.infra.local_sqlite import LocalSqlite
45+
46+
assert config.online_store.local is not None
47+
return LocalSqlite(config.online_store.local)
48+
else:
49+
raise ValueError(config)

‎sdk/python/feast/repo_config.py‎

Lines changed: 31 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,31 @@
1+
from pathlib import Path
2+
from typing import NamedTuple, Optional
3+
4+
import yaml
5+
from bindr import bind
6+
7+
8+
class LocalOnlineStoreConfig(NamedTuple):
9+
path: str
10+
11+
12+
class DatastoreOnlineStoreConfig(NamedTuple):
13+
project_id: str
14+
15+
16+
class OnlineStoreConfig(NamedTuple):
17+
datastore: Optional[DatastoreOnlineStoreConfig] = None
18+
local: Optional[LocalOnlineStoreConfig] = None
19+
20+
21+
class RepoConfig(NamedTuple):
22+
metadata_store: str
23+
project: str
24+
provider: str
25+
online_store: OnlineStoreConfig
26+
27+
28+
def load_repo_config(repo_path: Path) -> RepoConfig:
29+
with open(repo_path / "feature_store.yaml") as f:
30+
raw_config = yaml.safe_load(f)
31+
return bind(RepoConfig, raw_config)

0 commit comments

Comments
 (0)