From 39cac6887e53911e6b43b341d8894c8ed3d5ed80 Mon Sep 17 00:00:00 2001 From: mike0sv Date: Mon, 16 Nov 2020 19:34:03 +0300 Subject: [PATCH 1/9] S3 endpoint configuration #1169 Signed-off-by: mike0sv --- sdk/python/feast/client.py | 5 +++- sdk/python/feast/constants.py | 4 ++++ sdk/python/feast/loaders/ingest.py | 5 ++-- sdk/python/feast/staging/entities.py | 5 ++-- sdk/python/feast/staging/storage_client.py | 28 ++++++++++++++++++---- 5 files changed, 37 insertions(+), 10 deletions(-) diff --git a/sdk/python/feast/client.py b/sdk/python/feast/client.py index 60b320b9f06..cd0a396f624 100644 --- a/sdk/python/feast/client.py +++ b/sdk/python/feast/client.py @@ -824,7 +824,9 @@ def ingest( try: if issubclass(type(feature_table.batch_source), FileSource): file_url = feature_table.batch_source.file_options.file_url.rstrip("*") - _upload_to_file_source(file_url, with_partitions, dest_path) + _upload_to_file_source( + file_url, with_partitions, dest_path, self._config + ) if issubclass(type(feature_table.batch_source), BigQuerySource): bq_table_ref = feature_table.batch_source.bigquery_options.table_ref feature_table_timestamp_column = ( @@ -979,6 +981,7 @@ def get_historical_features( entity_source = stage_entities_to_fs( entity_source, staging_location=self._config.get(CONFIG_SPARK_STAGING_LOCATION), + config=self._config, ) if self._use_job_service: diff --git a/sdk/python/feast/constants.py b/sdk/python/feast/constants.py index 3f0614b097e..4bbaa233c7c 100644 --- a/sdk/python/feast/constants.py +++ b/sdk/python/feast/constants.py @@ -68,6 +68,8 @@ class AuthProvider(Enum): CONFIG_TIMEOUT_KEY = "timeout" CONFIG_MAX_WAIT_INTERVAL_KEY = "max_wait_interval" +CONFIG_S3_ENDPOINT_URL = "s3_endpoint_url" + # Spark Job Config CONFIG_SPARK_LAUNCHER = "spark_launcher" # standalone, dataproc, emr @@ -130,6 +132,8 @@ class AuthProvider(Enum): CONFIG_BATCH_FEATURE_REQUEST_WAIT_TIME_SECONDS_KEY: "600", CONFIG_TIMEOUT_KEY: "21600", CONFIG_MAX_WAIT_INTERVAL_KEY: "60", + # Endpoint URL for S3 storage_client + CONFIG_S3_ENDPOINT_URL: None, # Authentication Provider - Google OpenID/OAuth CONFIG_AUTH_PROVIDER: "google", CONFIG_SPARK_LAUNCHER: "dataproc", diff --git a/sdk/python/feast/loaders/ingest.py b/sdk/python/feast/loaders/ingest.py index dc87d5b32e5..dcda1b39b76 100644 --- a/sdk/python/feast/loaders/ingest.py +++ b/sdk/python/feast/loaders/ingest.py @@ -9,6 +9,7 @@ import pyarrow as pa from pyarrow import parquet as pq +from feast.config import Config from feast.staging.storage_client import get_staging_client GRPC_CONNECTION_TIMEOUT_DEFAULT = 3 # type: int @@ -173,7 +174,7 @@ def _read_table_from_source( def _upload_to_file_source( - file_url: str, with_partitions: bool, dest_path: str + file_url: str, with_partitions: bool, dest_path: str, config: Config ) -> None: """ Uploads data into a FileSource. Currently supports GCS, S3 and Local FS. @@ -184,7 +185,7 @@ def _upload_to_file_source( from urllib.parse import urlparse uri = urlparse(file_url) - staging_client = get_staging_client(uri.scheme) + staging_client = get_staging_client(uri.scheme, config) if with_partitions: for path in glob.glob(os.path.join(dest_path, "**/*")): diff --git a/sdk/python/feast/staging/entities.py b/sdk/python/feast/staging/entities.py index 8a4745fe24c..665cbcac13d 100644 --- a/sdk/python/feast/staging/entities.py +++ b/sdk/python/feast/staging/entities.py @@ -7,6 +7,7 @@ import pandas as pd +from feast.config import Config from feast.data_format import ParquetFormat from feast.data_source import BigQuerySource, FileSource from feast.staging.storage_client import get_staging_client @@ -18,7 +19,7 @@ def stage_entities_to_fs( - entity_source: pd.DataFrame, staging_location: str + entity_source: pd.DataFrame, staging_location: str, config: Config ) -> FileSource: """ Dumps given (entities) dataframe as parquet file and stage it to remote file storage (subdirectory of staging_location) @@ -26,7 +27,7 @@ def stage_entities_to_fs( :return: FileSource with remote destination path """ entity_staging_uri = urlparse(os.path.join(staging_location, str(uuid.uuid4()))) - staging_client = get_staging_client(entity_staging_uri.scheme) + staging_client = get_staging_client(entity_staging_uri.scheme, config) with tempfile.NamedTemporaryFile() as df_export_path: entity_source.to_parquet(df_export_path.name) bucket = ( diff --git a/sdk/python/feast/staging/storage_client.py b/sdk/python/feast/staging/storage_client.py index 1cb250a598b..1ea7162469d 100644 --- a/sdk/python/feast/staging/storage_client.py +++ b/sdk/python/feast/staging/storage_client.py @@ -24,6 +24,9 @@ from google.auth.exceptions import DefaultCredentialsError +from feast.config import Config +from feast.constants import CONFIG_S3_ENDPOINT_URL + GS = "gs" S3 = "s3" LOCAL_FILE = "file" @@ -144,7 +147,7 @@ class S3Client(AbstractStagingClient): Implementation of AbstractStagingClient for Aws S3 storage """ - def __init__(self): + def __init__(self, endpoint_url: str = None): try: import boto3 except ImportError: @@ -152,7 +155,7 @@ def __init__(self): "Install package boto3 for s3 staging support" "run ```pip install boto3```" ) - self.s3_client = boto3.client("s3") + self.s3_client = boto3.client("s3", endpoint_url=endpoint_url) def download_file(self, uri: ParseResult) -> IO[bytes]: """ @@ -275,21 +278,36 @@ def upload_file(self, local_path: str, bucket: str, remote_path: str): shutil.copy(local_path, dest_fpath) -storage_clients = {GS: GCSClient, S3: S3Client, LOCAL_FILE: LocalFSClient} +def _s3_client(config: Config = None): + return S3Client( + endpoint_url=config.get(CONFIG_S3_ENDPOINT_URL) if config is not None else None + ) + + +def _gcs_client(config: Config = None): + return GCSClient() + + +def _local_fs_client(config: Config = None): + return LocalFSClient() + + +storage_clients = {GS: _gcs_client, S3: _s3_client, LOCAL_FILE: _local_fs_client} -def get_staging_client(scheme): +def get_staging_client(scheme, config: Config = None): """ Initialization of a specific client object(GCSClient, S3Client etc.) Args: scheme (str): uri scheme: s3, gs or file + config (Config): additional configuration Returns: An object of concrete implementation of AbstractStagingClient """ try: - return storage_clients[scheme]() + return storage_clients[scheme](config) except ValueError: raise Exception( f"Could not identify file scheme {scheme}. Only gs://, file:// and s3:// are supported" From 01def7f061db13fd5c39e796a3b97d581ef0473b Mon Sep 17 00:00:00 2001 From: mike0sv Date: Tue, 17 Nov 2020 08:30:06 +0300 Subject: [PATCH 2/9] Add allow_no_value=True to ConfigParser Signed-off-by: mike0sv --- sdk/python/feast/config.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/sdk/python/feast/config.py b/sdk/python/feast/config.py index 1bbab4edbcf..ae79408ff2e 100644 --- a/sdk/python/feast/config.py +++ b/sdk/python/feast/config.py @@ -50,7 +50,7 @@ def _init_config(path: str): os.makedirs(os.path.dirname(config_dir), exist_ok=True) # Create the configuration file itself - config = ConfigParser(defaults=DEFAULTS) + config = ConfigParser(defaults=DEFAULTS, allow_no_value=True) if os.path.exists(path): config.read(path) From 6aa943fa67da52c314dd4099a198e7281ccb2143 Mon Sep 17 00:00:00 2001 From: mike0sv Date: Fri, 20 Nov 2020 19:46:21 +0300 Subject: [PATCH 3/9] New constants API defaults extraction Signed-off-by: mike0sv --- sdk/python/feast/client.py | 2 +- sdk/python/feast/config.py | 15 ++++++++++++--- sdk/python/feast/constants.py | 3 +++ sdk/python/feast/staging/storage_client.py | 10 ++++++---- 4 files changed, 22 insertions(+), 8 deletions(-) diff --git a/sdk/python/feast/client.py b/sdk/python/feast/client.py index fa4437e7ac4..0d7d689c5de 100644 --- a/sdk/python/feast/client.py +++ b/sdk/python/feast/client.py @@ -1006,7 +1006,7 @@ def get_historical_features( entity_source = stage_entities_to_fs( entity_source, staging_location=self._config.get(opt.SPARK_STAGING_LOCATION), - config=self._config + config=self._config, ) if self._use_job_service: diff --git a/sdk/python/feast/config.py b/sdk/python/feast/config.py index 32950025099..ba3ded76576 100644 --- a/sdk/python/feast/config.py +++ b/sdk/python/feast/config.py @@ -29,6 +29,7 @@ from feast.constants import ConfigOptions as opt _logger = logging.getLogger(__name__) +_UNSET = object() def _init_config(path: str): @@ -50,7 +51,7 @@ def _init_config(path: str): os.makedirs(os.path.dirname(config_dir), exist_ok=True) # Create the configuration file itself - config = ConfigParser(defaults=opt().defaults(), allow_no_value=True) + config = ConfigParser(allow_no_value=True) if os.path.exists(path): config.read(path) @@ -113,24 +114,32 @@ def __init__( self._options = {} if options and isinstance(options, dict): self._options = options + self._defaults = opt().defaults() self._config = config # type: ConfigParser self._path = path # type: str - def get(self, option): + def get(self, option, default=_UNSET): """ Returns a single configuration option as a string Args: option: Name of the option + default: Default value to return if option is not found Returns: String option that is returned """ + default = {option: default} if default is not _UNSET else {} return self._config.get( CONFIG_FILE_SECTION, option, - vars={**_get_feast_env_vars(), **self._options}, + vars={ + **default, + **self._defaults, + **_get_feast_env_vars(), + **self._options, + }, ) def getboolean(self, option): diff --git a/sdk/python/feast/constants.py b/sdk/python/feast/constants.py index 4d677bc38d8..8b1db7b76bd 100644 --- a/sdk/python/feast/constants.py +++ b/sdk/python/feast/constants.py @@ -125,6 +125,9 @@ class ConfigOptions(metaclass=ConfigMeta): #: Time to wait for historical feature requests before timing out. BATCH_FEATURE_REQUEST_WAIT_TIME_SECONDS: str = "600" + #: Endpoint URL for S3 storage_client + S3_ENDPOINT_URL: Optional[str] = None + #: Authentication Provider - Google OpenID/OAuth #: #: Options: "google" / "oauth" diff --git a/sdk/python/feast/staging/storage_client.py b/sdk/python/feast/staging/storage_client.py index 1ea7162469d..9d6c9f52a43 100644 --- a/sdk/python/feast/staging/storage_client.py +++ b/sdk/python/feast/staging/storage_client.py @@ -25,7 +25,7 @@ from google.auth.exceptions import DefaultCredentialsError from feast.config import Config -from feast.constants import CONFIG_S3_ENDPOINT_URL +from feast.constants import ConfigOptions as opt GS = "gs" S3 = "s3" @@ -279,9 +279,11 @@ def upload_file(self, local_path: str, bucket: str, remote_path: str): def _s3_client(config: Config = None): - return S3Client( - endpoint_url=config.get(CONFIG_S3_ENDPOINT_URL) if config is not None else None - ) + if config is None: + endpoint_url = None + else: + endpoint_url = config.get(opt.S3_ENDPOINT_URL, None) + return S3Client(endpoint_url=endpoint_url) def _gcs_client(config: Config = None): From caf1f6d4adbe56e6c8a2ff3e9b7e7b2d802387d3 Mon Sep 17 00:00:00 2001 From: mike0sv Date: Fri, 20 Nov 2020 20:07:59 +0300 Subject: [PATCH 4/9] fix for other types of get Signed-off-by: mike0sv --- sdk/python/feast/config.py | 33 +++++++++++++++++++++++++++------ 1 file changed, 27 insertions(+), 6 deletions(-) diff --git a/sdk/python/feast/config.py b/sdk/python/feast/config.py index ba3ded76576..297b6b15e64 100644 --- a/sdk/python/feast/config.py +++ b/sdk/python/feast/config.py @@ -142,52 +142,73 @@ def get(self, option, default=_UNSET): }, ) - def getboolean(self, option): + def getboolean(self, option, default=_UNSET): """ Returns a single configuration option as a boolean Args: option: Name of the option + default: Default value to return if option is not found Returns: Boolean option value that is returned """ + default = {option: default} if default is not _UNSET else {} return self._config.getboolean( CONFIG_FILE_SECTION, option, - vars={**_get_feast_env_vars(), **self._options}, + vars={ + **default, + **self._defaults, + **_get_feast_env_vars(), + **self._options, + }, ) - def getint(self, option): + def getint(self, option, default=_UNSET): """ Returns a single configuration option as an integer Args: option: Name of the option + default: Default value to return if option is not found Returns: Integer option value that is returned """ + default = {option: default} if default is not _UNSET else {} return self._config.getint( CONFIG_FILE_SECTION, option, - vars={**_get_feast_env_vars(), **self._options}, + vars={ + **default, + **self._defaults, + **_get_feast_env_vars(), + **self._options, + }, ) - def getfloat(self, option): + def getfloat(self, option, default=_UNSET): """ Returns a single configuration option as an integer Args: option: Name of the option + default: Default value to return if option is not found Returns: Float option value that is returned """ + default = {option: default} if default is not _UNSET else {} return self._config.getfloat( CONFIG_FILE_SECTION, option, - vars={**_get_feast_env_vars(), **self._options}, + vars={ + **default, + **self._defaults, + **_get_feast_env_vars(), + **self._options, + }, ) def set(self, option, value): From ca3ce661a1d4579cbca7d3700db49c8a14b1b8f3 Mon Sep 17 00:00:00 2001 From: mike0sv Date: Sat, 21 Nov 2020 09:18:45 +0300 Subject: [PATCH 5/9] return to the old logic and some testing Signed-off-by: mike0sv --- sdk/python/feast/config.py | 77 ++++++++++++--------------------- sdk/python/tests/test_config.py | 21 +++++++++ 2 files changed, 48 insertions(+), 50 deletions(-) diff --git a/sdk/python/feast/config.py b/sdk/python/feast/config.py index 297b6b15e64..c6fa8be0a48 100644 --- a/sdk/python/feast/config.py +++ b/sdk/python/feast/config.py @@ -51,7 +51,7 @@ def _init_config(path: str): os.makedirs(os.path.dirname(config_dir), exist_ok=True) # Create the configuration file itself - config = ConfigParser(allow_no_value=True) + config = ConfigParser(defaults=opt().defaults(), allow_no_value=True) if os.path.exists(path): config.read(path) @@ -59,9 +59,6 @@ def _init_config(path: str): if not config.has_section(CONFIG_FILE_SECTION): config.add_section(CONFIG_FILE_SECTION) - # Save the current configuration - config.write(open(path, "w")) - return config @@ -114,11 +111,19 @@ def __init__( self._options = {} if options and isinstance(options, dict): self._options = options - self._defaults = opt().defaults() self._config = config # type: ConfigParser self._path = path # type: str + def _get(self, option, default, get_method): + fallback = {} if default is _UNSET else {"fallback": default} + return get_method( + CONFIG_FILE_SECTION, + option, + vars={**_get_feast_env_vars(), **self._options,}, + **fallback, + ) + def get(self, option, default=_UNSET): """ Returns a single configuration option as a string @@ -130,17 +135,7 @@ def get(self, option, default=_UNSET): Returns: String option that is returned """ - default = {option: default} if default is not _UNSET else {} - return self._config.get( - CONFIG_FILE_SECTION, - option, - vars={ - **default, - **self._defaults, - **_get_feast_env_vars(), - **self._options, - }, - ) + return self._get(option, default, self._config.get) def getboolean(self, option, default=_UNSET): """ @@ -153,17 +148,7 @@ def getboolean(self, option, default=_UNSET): Returns: Boolean option value that is returned """ - default = {option: default} if default is not _UNSET else {} - return self._config.getboolean( - CONFIG_FILE_SECTION, - option, - vars={ - **default, - **self._defaults, - **_get_feast_env_vars(), - **self._options, - }, - ) + return self._get(option, default, self._config.getboolean) def getint(self, option, default=_UNSET): """ @@ -176,17 +161,7 @@ def getint(self, option, default=_UNSET): Returns: Integer option value that is returned """ - default = {option: default} if default is not _UNSET else {} - return self._config.getint( - CONFIG_FILE_SECTION, - option, - vars={ - **default, - **self._defaults, - **_get_feast_env_vars(), - **self._options, - }, - ) + return self._get(option, default, self._config.getint) def getfloat(self, option, default=_UNSET): """ @@ -199,17 +174,7 @@ def getfloat(self, option, default=_UNSET): Returns: Float option value that is returned """ - default = {option: default} if default is not _UNSET else {} - return self._config.getfloat( - CONFIG_FILE_SECTION, - option, - vars={ - **default, - **self._defaults, - **_get_feast_env_vars(), - **self._options, - }, - ) + return self._get(option, default, self._config.getfloat) def set(self, option, value): """ @@ -241,7 +206,12 @@ def save(self): Save the current configuration to disk. This does not include environmental variables or initialized options """ - self._config.write(open(self._path, "w")) + defaults = self._config.defaults() + try: + self._config._defaults = {} + self._config.write(open(self._path, "w")) + finally: + self._config._defaults = defaults def __str__(self): result = "" @@ -250,3 +220,10 @@ def __str__(self): for name, value in self._config.items(section_name): result += name + " = " + value + "\n" return result + + +if __name__ == "__main__": + from feast import Client + + c = Client() + c._config.getboolean(opt.ENABLE_AUTH) diff --git a/sdk/python/tests/test_config.py b/sdk/python/tests/test_config.py index 9ed34a736a2..3c2f66dd01a 100644 --- a/sdk/python/tests/test_config.py +++ b/sdk/python/tests/test_config.py @@ -103,6 +103,17 @@ def test_default_options(self): config = Config(path=path) assert config.get("CORE_URL") == "localhost:6565" + def test_defaults_are_not_written(self): + """ + default values are not written to config file + """ + fd, path = mkstemp() + config = Config(path=path) + config.set("option", "value") + config.save() + with open(path) as f: + assert f.read() == "[general]\noption = value\n\n" + def test_type_casting(self): """ Test type casting of strings to other types @@ -117,6 +128,16 @@ def test_type_casting(self): assert config.getfloat("FLOAT_VAR") == 1.0 assert config.getboolean("BOOLEAN_VAR") is True + def test_type_casting_of_defaults(self): + """ + default values are casted as expected + """ + fd, path = mkstemp() + config = Config(path=path) + assert isinstance(config.getboolean("enable_auth"), bool) + assert isinstance(config.getint("DATAPROC_EXECUTOR_INSTANCES"), int) + assert isinstance(config.getfloat("DATAPROC_EXECUTOR_INSTANCES"), float) + def test_set_value(self): """ Test type casting of strings to other types From 2684151c1f4a7bd657b4c514745f84b25075c897 Mon Sep 17 00:00:00 2001 From: mike0sv Date: Sat, 21 Nov 2020 09:29:38 +0300 Subject: [PATCH 6/9] oooopsie Signed-off-by: mike0sv --- sdk/python/feast/config.py | 9 +-------- 1 file changed, 1 insertion(+), 8 deletions(-) diff --git a/sdk/python/feast/config.py b/sdk/python/feast/config.py index c6fa8be0a48..788d257b9f3 100644 --- a/sdk/python/feast/config.py +++ b/sdk/python/feast/config.py @@ -120,7 +120,7 @@ def _get(self, option, default, get_method): return get_method( CONFIG_FILE_SECTION, option, - vars={**_get_feast_env_vars(), **self._options,}, + vars={**_get_feast_env_vars(), **self._options}, **fallback, ) @@ -220,10 +220,3 @@ def __str__(self): for name, value in self._config.items(section_name): result += name + " = " + value + "\n" return result - - -if __name__ == "__main__": - from feast import Client - - c = Client() - c._config.getboolean(opt.ENABLE_AUTH) From 1f181f76e25f7f48bd779a742192d59113d01872 Mon Sep 17 00:00:00 2001 From: mike0sv Date: Sun, 22 Nov 2020 05:23:16 +0300 Subject: [PATCH 7/9] remove DEFAULTS logic changes Signed-off-by: mike0sv --- sdk/python/feast/constants.py | 3 +++ sdk/python/feast/staging/storage_client.py | 4 ++-- 2 files changed, 5 insertions(+), 2 deletions(-) diff --git a/sdk/python/feast/constants.py b/sdk/python/feast/constants.py index 4d677bc38d8..8b1db7b76bd 100644 --- a/sdk/python/feast/constants.py +++ b/sdk/python/feast/constants.py @@ -125,6 +125,9 @@ class ConfigOptions(metaclass=ConfigMeta): #: Time to wait for historical feature requests before timing out. BATCH_FEATURE_REQUEST_WAIT_TIME_SECONDS: str = "600" + #: Endpoint URL for S3 storage_client + S3_ENDPOINT_URL: Optional[str] = None + #: Authentication Provider - Google OpenID/OAuth #: #: Options: "google" / "oauth" diff --git a/sdk/python/feast/staging/storage_client.py b/sdk/python/feast/staging/storage_client.py index 1ea7162469d..c873df94e65 100644 --- a/sdk/python/feast/staging/storage_client.py +++ b/sdk/python/feast/staging/storage_client.py @@ -25,7 +25,7 @@ from google.auth.exceptions import DefaultCredentialsError from feast.config import Config -from feast.constants import CONFIG_S3_ENDPOINT_URL +from feast.constants import ConfigOptions as opt GS = "gs" S3 = "s3" @@ -280,7 +280,7 @@ def upload_file(self, local_path: str, bucket: str, remote_path: str): def _s3_client(config: Config = None): return S3Client( - endpoint_url=config.get(CONFIG_S3_ENDPOINT_URL) if config is not None else None + endpoint_url=config.get(opt.S3_ENDPOINT_URL) if config is not None else None ) From 8e7dab45bfdae4e8ebd2c0c9e2b0c3fd139db42a Mon Sep 17 00:00:00 2001 From: mike0sv Date: Sun, 22 Nov 2020 05:31:50 +0300 Subject: [PATCH 8/9] reformat Signed-off-by: mike0sv --- sdk/python/feast/client.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/sdk/python/feast/client.py b/sdk/python/feast/client.py index fa4437e7ac4..0d7d689c5de 100644 --- a/sdk/python/feast/client.py +++ b/sdk/python/feast/client.py @@ -1006,7 +1006,7 @@ def get_historical_features( entity_source = stage_entities_to_fs( entity_source, staging_location=self._config.get(opt.SPARK_STAGING_LOCATION), - config=self._config + config=self._config, ) if self._use_job_service: From 081288d97157bbd2f55fa1fc29cfe0e718c34e72 Mon Sep 17 00:00:00 2001 From: mike0sv Date: Sun, 22 Nov 2020 11:42:27 +0300 Subject: [PATCH 9/9] _upload_to_file_source docs Signed-off-by: mike0sv --- sdk/python/feast/loaders/ingest.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/sdk/python/feast/loaders/ingest.py b/sdk/python/feast/loaders/ingest.py index 6d39519cab8..67c8f5f726b 100644 --- a/sdk/python/feast/loaders/ingest.py +++ b/sdk/python/feast/loaders/ingest.py @@ -174,6 +174,9 @@ def _upload_to_file_source( Args: file_url: file url of FileSource defined for FeatureTable + with_partitions: whether to treat dest_path as dir with partitioned table + dest_path: path to file or dir to be uploaded + config: Config instance to configure FileSource """ from urllib.parse import urlparse