Add support for DynamoDB and S3 registry

Makefile

@@ -41,6 +41,9 @@ install-ci-dependencies: install-python-ci-dependencies install-go-ci-dependenci
 install-python-ci-dependencies:
 	pip install -e "sdk/python[ci]"
 
+install-python-aws-dependencies:
+	pip install -e "sdk/python[aws]"
+
 package-protos:
 	cp -r ${ROOT_DIR}/protos ${ROOT_DIR}/sdk/python/feast/protos
 

sdk/python/feast/cli.py

@@ -280,7 +280,7 @@ def materialize_incremental_command(ctx: click.Context, end_ts: str, views: List
 @click.option(
     "--template",
     "-t",
-    type=click.Choice(["local", "gcp"], case_sensitive=False),
+    type=click.Choice(["local", "gcp", "aws"], case_sensitive=False),
     help="Specify a template for the created project",
     default="local",
 )

sdk/python/feast/errors.py

@@ -40,6 +40,26 @@ def __init__(self, name, project=None):
             super().__init__(f"Feature table {name} does not exist")
 
 
+class FeatureBucketNotExist(FeastObjectNotFoundException):
+    def __init__(self, bucket, project=None):
+        if project:
+            super().__init__(
+                f"Feature bucket {bucket} does not exist in project {project}"
+            )
+        else:
+            super().__init__(f"Feature bucket {bucket} does not exist")
+
+
+class FeatureBucketForbiddenAccess(FeastObjectNotFoundException):
+    def __init__(self, bucket, project=None):
+        if project:
+            super().__init__(
+                f"Private Registry Bucket {bucket} forbidden Access in project {project}"
+            )
+        else:
+            super().__init__(f"Private Registry Bucket {bucket} forbidden Access")
+
+
 class FeastProviderLoginError(Exception):
     """Error class that indicates a user has not authenticated with their provider."""
 

sdk/python/feast/infra/aws_provider.py

@@ -0,0 +1,136 @@
+from datetime import datetime
+from typing import Any, Callable, Dict, List, Optional, Sequence, Tuple, Union
+
+import pandas
+from tqdm import tqdm
+
+from feast import FeatureTable, utils
+from feast.entity import Entity
+from feast.feature_view import FeatureView
+from feast.infra.offline_stores.helpers import get_offline_store_from_config
+from feast.infra.online_stores.helpers import get_online_store_from_config
+from feast.infra.provider import (
+    Provider,
+    RetrievalJob,
+    _convert_arrow_to_proto,
+    _get_column_names,
+    _run_field_mapping,
+)
+from feast.protos.feast.types.EntityKey_pb2 import EntityKey as EntityKeyProto
+from feast.protos.feast.types.Value_pb2 import Value as ValueProto
+from feast.registry import Registry
+from feast.repo_config import RepoConfig
+
+
+class AwsProvider(Provider):
+    def __init__(self, config: RepoConfig):
+        self.repo_config = config
+        self.offline_store = get_offline_store_from_config(config.offline_store)
+        self.online_store = get_online_store_from_config(config.online_store)
+
+    def update_infra(
+        self,
+        project: str,
+        tables_to_delete: Sequence[Union[FeatureTable, FeatureView]],
+        tables_to_keep: Sequence[Union[FeatureTable, FeatureView]],
+        entities_to_delete: Sequence[Entity],
+        entities_to_keep: Sequence[Entity],
+        partial: bool,
+    ):
+        self.online_store.update(
+            config=self.repo_config,
+            tables_to_delete=tables_to_delete,
+            tables_to_keep=tables_to_keep,
+            entities_to_keep=entities_to_keep,
+            entities_to_delete=entities_to_delete,
+            partial=partial,
+        )
+
+    def teardown_infra(
+        self,
+        project: str,
+        tables: Sequence[Union[FeatureTable, FeatureView]],
+        entities: Sequence[Entity],
+    ) -> None:
+        self.online_store.teardown(self.repo_config, tables, entities)
+
+    def online_write_batch(
+        self,
+        config: RepoConfig,
+        table: Union[FeatureTable, FeatureView],
+        data: List[
+            Tuple[EntityKeyProto, Dict[str, ValueProto], datetime, Optional[datetime]]
+        ],
+        progress: Optional[Callable[[int], Any]],
+    ) -> None:
+        self.online_store.online_write_batch(config, table, data, progress)
+
+    def online_read(
+        self,
+        config: RepoConfig,
+        table: Union[FeatureTable, FeatureView],
+        entity_keys: List[EntityKeyProto],
+        requested_features: List[str] = None,
+    ) -> List[Tuple[Optional[datetime], Optional[Dict[str, ValueProto]]]]:
+        result = self.online_store.online_read(config, table, entity_keys)
+
+        return result
+
+    def materialize_single_feature_view(
+        self,
+        feature_view: FeatureView,
+        start_date: datetime,
+        end_date: datetime,
+        registry: Registry,
+        project: str,
+        tqdm_builder: Callable[[int], tqdm],
+    ) -> None:
+        entities = []
+        for entity_name in feature_view.entities:
+            entities.append(registry.get_entity(entity_name, project))
+
+        (
+            join_key_columns,
+            feature_name_columns,
+            event_timestamp_column,
+            created_timestamp_column,
+        ) = _get_column_names(feature_view, entities)
+
+        start_date = utils.make_tzaware(start_date)
+        end_date = utils.make_tzaware(end_date)
+
+        table = self.offline_store.pull_latest_from_table_or_query(
+            data_source=feature_view.input,
+            join_key_columns=join_key_columns,
+            feature_name_columns=feature_name_columns,
+            event_timestamp_column=event_timestamp_column,
+            created_timestamp_column=created_timestamp_column,
+            start_date=start_date,
+            end_date=end_date,
+        )
+
+        if feature_view.input.field_mapping is not None:
+            table = _run_field_mapping(table, feature_view.input.field_mapping)
+
+        join_keys = [entity.join_key for entity in entities]
+        rows_to_write = _convert_arrow_to_proto(table, feature_view, join_keys)
+
+        with tqdm_builder(len(rows_to_write)) as pbar:
+            self.online_write_batch(
+                self.repo_config, feature_view, rows_to_write, lambda x: pbar.update(x)
+            )
+
+        feature_view.materialization_intervals.append((start_date, end_date))
+        registry.apply_feature_view(feature_view, project)
+
+    @staticmethod
+    def get_historical_features(
+        config: RepoConfig,
+        feature_views: List[FeatureView],
+        feature_refs: List[str],
+        entity_df: Union[pandas.DataFrame, str],
+        registry: Registry,
+        project: str,
+    ) -> RetrievalJob:
+        # TODO implement me
+        pass

sdk/python/feast/infra/online_stores/dynamodb.py

@@ -0,0 +1,186 @@
+# Copyright 2021 The Feast Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+from datetime import datetime
+from typing import Any, Callable, Dict, List, Optional, Sequence, Tuple, Union
+
+import mmh3
+from pydantic import PositiveInt, StrictStr
+from pydantic.typing import Literal
+
+from feast import Entity, FeatureTable, FeatureView, utils
+from feast.infra.key_encoding_utils import serialize_entity_key
+from feast.infra.online_stores.online_store import OnlineStore
+from feast.protos.feast.types.EntityKey_pb2 import EntityKey as EntityKeyProto
+from feast.protos.feast.types.Value_pb2 import Value as ValueProto
+from feast.repo_config import FeastConfigBaseModel, RepoConfig
+
+try:
+    import boto3
+    from botocore.exceptions import ClientError
+except ImportError as e:
+    from feast.errors import FeastExtrasDependencyImportError
+
+    raise FeastExtrasDependencyImportError("aws", str(e))
+
+
+class DynamoDbOnlineStoreConfig(FeastConfigBaseModel):
+    """Online store config for DynamoDB store"""
+
+    type: Literal["dynamodb"] = "dynamodb"
+    """Online store type selector"""
+
+    rcu: Optional[PositiveInt] = 5
+    """ Read capacity unit """
+
+    wcu: Optional[PositiveInt] = 5
+    """ Write capacity unit """
+
+    region_name: Optional[StrictStr] = None
+    """ AWS Region Name """
+
+
+class DynamoDbOnlineStore(OnlineStore):
+    def _initialize_dynamodb(self, online_config: DynamoDbOnlineStoreConfig):
+        return boto3.resource("dynamodb", region_name=online_config.region_name)
+
+    def update(
+        self,
+        config: RepoConfig,
+        tables_to_delete: Sequence[Union[FeatureTable, FeatureView]],
+        tables_to_keep: Sequence[Union[FeatureTable, FeatureView]],
+        entities_to_delete: Sequence[Entity],
+        entities_to_keep: Sequence[Entity],
+        partial: bool,
+    ):
+        online_config = config.online_store
+        assert isinstance(online_config, DynamoDbOnlineStoreConfig)
+        dynamodb = self._initialize_dynamodb(online_config)
+
+        for table_name in tables_to_keep:
+            table = None
+            try:
+                table = dynamodb.create_table(
+                    TableName=table_name.name,
+                    KeySchema=[
+                        {"AttributeName": "Row", "KeyType": "HASH"},
+                        {"AttributeName": "Project", "KeyType": "RANGE"},
+                    ],
+                    AttributeDefinitions=[
+                        {"AttributeName": "Row", "AttributeType": "S"},
+                        {"AttributeName": "Project", "AttributeType": "S"},
+                    ],
+                    ProvisionedThroughput={
+                        "ReadCapacityUnits": online_config.rcu,
+                        "WriteCapacityUnits": online_config.wcu,
+                    },
+                )
+                table.meta.client.get_waiter("table_exists").wait(
+                    TableName=table_name.name
+                )
+            except ClientError as ce:
+                print(ce)
+                if ce.response["Error"]["Code"] == "ResourceNotFoundException":
+                    table = dynamodb.Table(table_name.name)
+
+        for table_name in tables_to_delete:
+            table = dynamodb.Table(table_name.name)
+            table.delete()
+
+    def teardown(
+        self,
+        config: RepoConfig,
+        tables: Sequence[Union[FeatureTable, FeatureView]],
+        entities: Sequence[Entity],
+    ):
+        online_config = config.online_store
+        assert isinstance(online_config, DynamoDbOnlineStoreConfig)
+        dynamodb = self._initialize_dynamodb(online_config)
+
+        for table_name in tables:
+            try:
+                table = dynamodb.Table(table_name)
+                table.delete()
+            except Exception as e:
+                print(str(e))
+
+    def online_write_batch(
+        self,
+        config: RepoConfig,
+        table: Union[FeatureTable, FeatureView],
+        data: List[
+            Tuple[EntityKeyProto, Dict[str, ValueProto], datetime, Optional[datetime]]
+        ],
+        progress: Optional[Callable[[int], Any]],
+    ) -> None:
+        online_config = config.online_store
+        assert isinstance(online_config, DynamoDbOnlineStoreConfig)
+        dynamodb = self._initialize_dynamodb(online_config)
+
+        table_instance = dynamodb.Table(table.name)
+        with table_instance.batch_writer() as batch:
+            for entity_key, features, timestamp, created_ts in data:
+                document_id = compute_datastore_entity_id(entity_key)  # TODO check id
+                # TODO compression encoding
+                batch.put_item(
+                    Item={
+                        "Row": document_id,  # PartitionKey
+                        "Project": config.project,  # SortKey
+                        "event_ts": str(utils.make_tzaware(timestamp)),
+                        "values": {
+                            k: v.SerializeToString()
+                            for k, v in features.items()  # Serialized Features
+                        },
+                    }
+                )
+
+    def online_read(
+        self,
+        config: RepoConfig,
+        table: Union[FeatureTable, FeatureView],
+        entity_keys: List[EntityKeyProto],
+        requested_features: Optional[List[str]] = None,
+    ) -> List[Tuple[Optional[datetime], Optional[Dict[str, ValueProto]]]]:
+        online_config = config.online_store
+        assert isinstance(online_config, DynamoDbOnlineStoreConfig)
+        dynamodb = self._initialize_dynamodb(online_config)
+
+        result: List[Tuple[Optional[datetime], Optional[Dict[str, ValueProto]]]] = []
+        for entity_key in entity_keys:
+            table_instace = dynamodb.Table(table.name)
+            document_id = compute_datastore_entity_id(entity_key)  # TODO check id
+            response = table_instace.get_item(
+                Key={"Row": document_id, "Project": config.project}
+            )
+            value = response["Item"]
+
+            if value is not None:
+                res = {}
+                for feature_name, value_bin in value["values"].items():
+                    val = ValueProto()
+                    val.ParseFromString(value_bin.value)
+                    res[feature_name] = val
+                result.append((value["event_ts"], res))
+            else:
+                result.append((None, None))
+        return result
+
+
+def compute_datastore_entity_id(entity_key: EntityKeyProto) -> str:
+    """
+    Compute Datastore Entity id given Feast Entity Key.
+
+    Remember that Datastore Entity is a concept from the Datastore data model, that has nothing to
+    do with the Entity concept we have in Feast.
+    """
+    return mmh3.hash_bytes(serialize_entity_key(entity_key)).hex()

sdk/python/feast/infra/provider.py

@@ -144,6 +144,10 @@ def get_provider(config: RepoConfig, repo_path: Path) -> Provider:
             from feast.infra.gcp import GcpProvider
 
             return GcpProvider(config)
+        elif config.provider == "aws":
+            from feast.infra.aws_provider import AwsProvider
+
+            return AwsProvider(config)
         elif config.provider == "local":
             from feast.infra.local import LocalProvider