From 43a2af8f56ae466c188166aba7842f9bf5fe93a8 Mon Sep 17 00:00:00 2001 From: Blair Chen Date: Mon, 31 Oct 2022 14:54:19 +0800 Subject: [PATCH] Revert "Expose 'timePartitionPattern' in Python API [ WIP ] (#714)" This reverts commit 3070a863d69f767658ca4adc819f6a3aac71ea18. --- feathr_project/feathr/definition/source.py | 21 +------ .../spark_provider/_synapse_submission.py | 2 - feathr_project/test/test_azure_spark_e2e.py | 36 +----------- feathr_project/test/test_feature_anchor.py | 58 +------------------ feathr_project/test/test_fixture.py | 28 --------- 5 files changed, 3 insertions(+), 142 deletions(-) diff --git a/feathr_project/feathr/definition/source.py b/feathr_project/feathr/definition/source.py index 3ca8925df..b9721a1a5 100644 --- a/feathr_project/feathr/definition/source.py +++ b/feathr_project/feathr/definition/source.py @@ -100,30 +100,14 @@ class HdfsSource(Source): - `epoch` (seconds since epoch), for example `1647737463` - `epoch_millis` (milliseconds since epoch), for example `1647737517761` - Any date formats supported by [SimpleDateFormat](https://docs.oracle.com/javase/8/docs/api/java/text/SimpleDateFormat.html). - registry_tags: A dict of (str, str) that you can pass to feature registry for better organization. For example, you can use {"deprecated": "true"} to indicate this source is deprecated, etc. - time_partition_pattern(Optional[str]): Format of the time partitioned feature data. e.g. yyyy/MM/DD. All formats supported in dateTimeFormatter. - config: - timeSnapshotHdfsSource: - { - location: - { - path: "/data/somePath/daily" - } - timePartitionPattern: "yyyy/MM/dd" - } - Given the above HDFS path: /data/somePath/daily, - then the expectation is that the following sub directorie(s) should exist: - /data/somePath/daily/{yyyy}/{MM}/{dd} - """ - def __init__(self, name: str, path: str, preprocessing: Optional[Callable] = None, event_timestamp_column: Optional[str] = None, timestamp_format: Optional[str] = "epoch", registry_tags: Optional[Dict[str, str]] = None, time_partition_pattern: Optional[str] = None) -> None: + def __init__(self, name: str, path: str, preprocessing: Optional[Callable] = None, event_timestamp_column: Optional[str] = None, timestamp_format: Optional[str] = "epoch", registry_tags: Optional[Dict[str, str]] = None) -> None: super().__init__(name, event_timestamp_column, timestamp_format, registry_tags=registry_tags) self.path = path self.preprocessing = preprocessing - self.time_partition_pattern = time_partition_pattern if path.startswith("http"): logger.warning( "Your input path {} starts with http, which is not supported. Consider using paths starting with wasb[s]/abfs[s]/s3.", path) @@ -132,9 +116,6 @@ def to_feature_config(self) -> str: tm = Template(""" {{source.name}}: { location: {path: "{{source.path}}"} - {% if source.time_partition_pattern %} - timePartitionPattern: "{{source.time_partition_pattern}}" - {% endif %} {% if source.event_timestamp_column %} timeWindowParameters: { timestampColumn: "{{source.event_timestamp_column}}" diff --git a/feathr_project/feathr/spark_provider/_synapse_submission.py b/feathr_project/feathr/spark_provider/_synapse_submission.py index 5b55a3b06..010444715 100644 --- a/feathr_project/feathr/spark_provider/_synapse_submission.py +++ b/feathr_project/feathr/spark_provider/_synapse_submission.py @@ -432,8 +432,6 @@ def download_file(self, target_adls_directory: str, local_dir_cache: str): # returns the paths to all the files in the target director in ADLS # get all the paths that are not under a directory - test_paths = self.file_system_client.get_paths( - path=parse_result.path, recursive=False) result_paths = [basename(file_path.name) for file_path in self.file_system_client.get_paths( path=parse_result.path, recursive=False) if not file_path.is_directory] diff --git a/feathr_project/test/test_azure_spark_e2e.py b/feathr_project/test/test_azure_spark_e2e.py index e82e0efe9..9c4ab8c5a 100644 --- a/feathr_project/test/test_azure_spark_e2e.py +++ b/feathr_project/test/test_azure_spark_e2e.py @@ -20,7 +20,7 @@ from feathr import ValueType from feathr.utils.job_utils import get_result_df from feathrcli.cli import init -from test_fixture import (basic_test_setup, get_online_test_table_name, time_partition_pattern_test_setup) +from test_fixture import (basic_test_setup, get_online_test_table_name) from test_utils.constants import Constants # make sure you have run the upload feature script before running these tests @@ -58,40 +58,6 @@ def test_feathr_materialize_to_offline(): res_df = get_result_df(client, "avro", output_path + "/df0/daily/2020/05/20") assert res_df.shape[0] > 0 -def test_feathr_materialize_with_time_partition_pattern(): - """ - Test FeathrClient() using HdfsSource with 'timePartitionPattern'. - """ - test_workspace_dir = Path( - __file__).parent.resolve() / "test_user_workspace" - # os.chdir(test_workspace_dir) - - client: FeathrClient = time_partition_pattern_test_setup(os.path.join(test_workspace_dir, "feathr_config.yaml")) - - backfill_time = BackfillTime(start=datetime( - 2020, 5, 20), end=datetime(2020, 5, 20), step=timedelta(days=1)) - - now = datetime.now() - if client.spark_runtime == 'databricks': - output_path = ''.join(['dbfs:/feathrazure_cijob_materialize_offline_','_', str(now.minute), '_', str(now.second), ""]) - else: - output_path = ''.join(['abfss://feathrazuretest3fs@feathrazuretest3storage.dfs.core.windows.net/demo_data/feathrazure_cijob_materialize_offline_','_', str(now.minute), '_', str(now.second), ""]) - offline_sink = HdfsSink(output_path=output_path) - settings = MaterializationSettings("nycTaxiTable", - sinks=[offline_sink], - feature_names=[ - "f_location_avg_fare", "f_location_max_fare"], - backfill_time=backfill_time) - client.materialize_features(settings) - # assuming the job can successfully run; otherwise it will throw exception - client.wait_job_to_finish(timeout_sec=Constants.SPARK_JOB_TIMEOUT_SECONDS) - - # download result and just assert the returned result is not empty - # by default, it will write to a folder appended with date - res_df = get_result_df(client, "avro", output_path + "/df0/daily/2020/05/20") - assert res_df.shape[0] > 0 - - def test_feathr_online_store_agg_features(): """ Test FeathrClient() get_online_features and batch_get can get data correctly. diff --git a/feathr_project/test/test_feature_anchor.py b/feathr_project/test/test_feature_anchor.py index 1fabeec67..d5e6701b8 100644 --- a/feathr_project/test/test_feature_anchor.py +++ b/feathr_project/test/test_feature_anchor.py @@ -167,60 +167,4 @@ def test_agg_anchor_to_config(): } } """ - assert ''.join(agg_anchor.to_feature_config().split()) == ''.join(expected_agg_feature_config.split()) - -def test_time_partition_to_config(): - batch_source = HdfsSource(name="testTimePartitionSource", - path="abfss://public@azurefeathrstorage.blob.core.windows.net/sample_data/time_partition_pattern", - time_partition_pattern="yyyy/MM/dd" - ) - key = TypedKey(key_column="key0", - key_column_type=ValueType.INT32) - agg_features = [ - Feature(name="f_loc_avg", - key=[key], - feature_type=FLOAT, - transform="f_location_avg_fare"), - Feature(name="f_loc_max", - feature_type=FLOAT, - key=[key], - transform="f_location_max_fare"), - ] - agg_anchor = FeatureAnchor(name="testTimePartitionFeaturesSource", - source=batch_source, - features=agg_features) - expected_time_partition_config = """ - anchors: { - testTimePartitionFeatures: { - source: testTimePartitionSource - key.sqlExpr: [key0] - features: { - f_loc_avg: { - def.sqlExpr: "f_location_avg_fare" - type: { - type: TENSOR - tensorCategory: DENSE - dimensionType: [] - valType: FLOAT - } - } - f_loc_max: { - def.sqlExpr: "f_location_max_fare" - type: { - type: TENSOR - tensorCategory: DENSE - dimensionType: [] - valType: FLOAT - } - } - } - } - } - sources: { - testTimePartitionSource: { - location: {path: "abfss://public@azurefeathrstorage.blob.core.windows.net/sample_data/time_partition_pattern"} - timePartitionPattern: "yyyy/MM/dd" - } - } - """ - assert ''.join(agg_anchor.to_feature_config().split()) == ''.join(expected_time_partition_config.split()) \ No newline at end of file + assert ''.join(agg_anchor.to_feature_config().split()) == ''.join(expected_agg_feature_config.split()) \ No newline at end of file diff --git a/feathr_project/test/test_fixture.py b/feathr_project/test/test_fixture.py index 637e74202..c048eff7c 100644 --- a/feathr_project/test/test_fixture.py +++ b/feathr_project/test/test_fixture.py @@ -88,34 +88,6 @@ def basic_test_setup(config_path: str): return client -def time_partition_pattern_test_setup(config_path: str): - now = datetime.now() - # set workspace folder by time; make sure we don't have write conflict if there are many CI tests running - os.environ['SPARK_CONFIG__DATABRICKS__WORK_DIR'] = ''.join(['dbfs:/feathrazure_cijob','_', str(now.minute), '_', str(now.second), '_', str(now.microsecond)]) - os.environ['SPARK_CONFIG__AZURE_SYNAPSE__WORKSPACE_DIR'] = ''.join(['abfss://feathrazuretest3fs@feathrazuretest3storage.dfs.core.windows.net/feathr_github_ci','_', str(now.minute), '_', str(now.second) ,'_', str(now.microsecond)]) - client = FeathrClient(config_path=config_path) - batch_source = HdfsSource(name="testTimePartitionSource", - path="wasbs://public@azurefeathrstorage.blob.core.windows.net/sample_data/time_partition_pattern/daily", - time_partition_pattern="yyyy/MM/dd" - ) - key = TypedKey(key_column="key0", - key_column_type=ValueType.INT32) - agg_features = [ - Feature(name="f_loc_avg", - key=[key], - feature_type=FLOAT, - transform="f_location_avg_fare"), - Feature(name="f_loc_max", - feature_type=FLOAT, - key=[key], - transform="f_location_max_fare"), - ] - - agg_anchor = FeatureAnchor(name="testTimePartitionFeatures", - source=batch_source, - features=agg_features) - client.build_features(anchor_list=[agg_anchor]) - return client def snowflake_test_setup(config_path: str): now = datetime.now()