From 582cf2b820d9776e0a50f216b5a261443dde6485 Mon Sep 17 00:00:00 2001 From: Hangfei Lin Date: Fri, 20 May 2022 21:09:24 -0700 Subject: [PATCH 1/9] [feathr] Add API to materialize features to offline store --- docs/concepts/feature-generation.md | 19 ++++++++- feathr_project/feathr/__init__.py | 2 +- feathr_project/feathr/sink.py | 41 ++++++++++++++++++- .../test/test_azure_snowflake_e2e.py | 10 ++--- feathr_project/test/test_azure_spark_e2e.py | 32 ++++++++++++++- .../test/test_feature_materialization.py | 37 ++++++++++++++--- 6 files changed, 125 insertions(+), 16 deletions(-) diff --git a/docs/concepts/feature-generation.md b/docs/concepts/feature-generation.md index 7e18557fe..d264c1b9d 100644 --- a/docs/concepts/feature-generation.md +++ b/docs/concepts/feature-generation.md @@ -5,10 +5,10 @@ parent: Feathr Concepts --- # Feature Generation +User could utilize feature generation to pre-compute and materialize pre-defined features to online and/or offline storage. This is a common practice when the feature transformation is computation intensive. ## Generating Features to Online Store - -User could utilize feature generation to pre-compute and materialize pre-defined features to online and/or offline storage. This is a common practice when the feature transformation is computation intensive. For example: +When we need to serve the models online, we also need to serve the features online. We provide APIs to generate features to online storage for future consumption. For example: ```python client = FeathrClient() redisSink = RedisSink(table_name="nycTaxiDemoFeature") @@ -49,3 +49,18 @@ res = client.get_online_features('nycTaxiDemoFeature', '265', [ ([client.get_online_features API doc](https://feathr.readthedocs.io/en/latest/feathr.html#feathr.client.FeathrClient.get_online_features)) After we finish running the materialization job, we can get the online features by querying the feature name, with the corresponding keys. In the example above, we query the online features called `f_location_avg_fare` and `f_location_max_fare`, and query with a key `265` (which is the location ID). + +## Generating Features to Offline Store + +User could utilize feature generation to pre-compute and materialize pre-defined features to offline storage. This is a common practice when the feature transformation is computation intensive and can be re-used. For example: +```python +client = FeathrClient() +offlineSink = OfflineSink(output_path="abfss://path/to/offline/hdfs_test.avro") +# Materialize two features into a Offline store. +settings = MaterializationSettings("nycTaxiMaterializationJob", + sinks=[offlineSink], + feature_names=["f_location_avg_fare", "f_location_max_fare"]) +client.materialize_features(settings) +``` +([MaterializationSettings API doc](https://feathr.readthedocs.io/en/latest/feathr.html#feathr.materialization_settings.MaterializationSettings), +[RedisSink API doc](https://feathr.readthedocs.io/en/latest/feathr.html#feathr.sink.RedisSink)) diff --git a/feathr_project/feathr/__init__.py b/feathr_project/feathr/__init__.py index 7e74ef90d..6957692ae 100644 --- a/feathr_project/feathr/__init__.py +++ b/feathr_project/feathr/__init__.py @@ -8,7 +8,7 @@ from .transformation import * from .typed_key import * from .materialization_settings import (BackfillTime, MaterializationSettings) -from .sink import RedisSink +from .sink import RedisSink, OfflineSink from .query_feature_list import FeatureQuery from .lookup_feature import LookupFeature from .aggregation import Aggregation diff --git a/feathr_project/feathr/sink.py b/feathr_project/feathr/sink.py index 75b5b7750..d4967755c 100644 --- a/feathr_project/feathr/sink.py +++ b/feathr_project/feathr/sink.py @@ -39,4 +39,43 @@ def to_feature_config(self) -> str: } """) msg = tm.render(source=self) - return msg \ No newline at end of file + return msg + + +# Sample generated HOCON config: +# operational: { +# name: testFeatureGen +# endTime: 2019-05-01 +# endTimeFormat: "yyyy-MM-dd" +# resolution: DAILY +# output:[ +# { +# name: HDFS +# params: { +# path: "/user/featureGen/hdfsResult/" +# } +# } +# ] +# } +# features: [mockdata_a_ct_gen, mockdata_a_sample_gen] +class OfflineSink(Sink): + """Offline Hadoop-compatible sink use to store offline feature data, can be used in batch job. + + Attributes: + table_name: output table name + """ + def __init__(self, output_path: str) -> None: + self.output_path = output_path + + def to_feature_config(self) -> str: + """Produce the config used in feature materialization""" + tm = Template(""" + { + name: HDFS + params: { + path: "{{sink.output_path}}" + } + } + """) + hocon_config = tm.render(sink=self) + return hocon_config \ No newline at end of file diff --git a/feathr_project/test/test_azure_snowflake_e2e.py b/feathr_project/test/test_azure_snowflake_e2e.py index b1f6785ca..7fff54a74 100644 --- a/feathr_project/test/test_azure_snowflake_e2e.py +++ b/feathr_project/test/test_azure_snowflake_e2e.py @@ -18,7 +18,7 @@ def test_feathr_online_store_agg_features(): Test FeathrClient() get_online_features and batch_get can get feature data correctly. """ test_workspace_dir = Path(__file__).parent.resolve() / "test_user_workspace" - + client = snowflake_test_setup(os.path.join(test_workspace_dir, "feathr_config.yaml")) online_test_table = get_online_test_table_name("snowflakeSampleDemoFeature") @@ -55,8 +55,8 @@ def test_feathr_get_offline_features(): Test get_offline_features() can get feature data from Snowflake source correctly. """ test_workspace_dir = Path(__file__).parent.resolve() / "test_user_workspace" - - + + client = snowflake_test_setup(os.path.join(test_workspace_dir, "feathr_config.yaml")) call_sk_id = TypedKey(key_column="CC_CALL_CENTER_SK", key_column_type=ValueType.INT32, @@ -69,14 +69,14 @@ def test_feathr_get_offline_features(): settings = ObservationSettings( observation_path='jdbc:snowflake://dqllago-ol19457.snowflakecomputing.com/?user=feathrintegration&sfWarehouse' '=COMPUTE_WH&dbtable=CALL_CENTER&sfDatabase=SNOWFLAKE_SAMPLE_DATA&sfSchema=TPCDS_SF10TCL') - + now = datetime.now() # set output folder based on different runtime if client.spark_runtime == 'databricks': output_path = ''.join(['dbfs:/feathrazure_cijob_snowflake','_', str(now.minute), '_', str(now.second), ".avro"]) else: output_path = ''.join(['abfss://feathrazuretest3fs@feathrazuretest3storage.dfs.core.windows.net/demo_data/snowflake_output','_', str(now.minute), '_', str(now.second), ".avro"]) - + client.get_offline_features(observation_settings=settings, feature_query=feature_query, output_path=output_path) diff --git a/feathr_project/test/test_azure_spark_e2e.py b/feathr_project/test/test_azure_spark_e2e.py index c22765119..f3c9cdf98 100644 --- a/feathr_project/test/test_azure_spark_e2e.py +++ b/feathr_project/test/test_azure_spark_e2e.py @@ -10,7 +10,7 @@ from feathr import (BackfillTime, MaterializationSettings) from feathr import FeatureQuery from feathr import ObservationSettings -from feathr import RedisSink +from feathr import RedisSink, OfflineSink from feathr import TypedKey from feathrcli.cli import init import pytest @@ -18,6 +18,34 @@ from test_fixture import (basic_test_setup, get_online_test_table_name) # make sure you have run the upload feature script before running these tests # the feature configs are from feathr_project/data/feathr_user_workspace +def test_feathr_materialize_to_offline(): + """ + Test FeathrClient() OfflineSink. + """ + + online_test_table = get_online_test_table_name("nycTaxiCITable") + test_workspace_dir = Path( + __file__).parent.resolve() / "test_user_workspace" + # os.chdir(test_workspace_dir) + + client = basic_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)) + offline_sink = OfflineSink(output_path="abfss://feathrazuretest3fs@feathrazuretest3storage.dfs.core.windows.net/demo_data/output/hdfs_test.avro") + 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=900) + + # download result and just assert the returned result is not empty + res_df = get_result_df(client) + 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. @@ -69,7 +97,7 @@ def test_feathr_online_store_non_agg_features(): test_workspace_dir = Path( __file__).parent.resolve() / "test_user_workspace" client = basic_test_setup(os.path.join(test_workspace_dir, "feathr_config.yaml")) - + online_test_table = get_online_test_table_name('nycTaxiCITable') backfill_time = BackfillTime(start=datetime( 2020, 5, 20), end=datetime(2020, 5, 20), step=timedelta(days=1)) diff --git a/feathr_project/test/test_feature_materialization.py b/feathr_project/test/test_feature_materialization.py index 74a8c2f08..c95ef2497 100644 --- a/feathr_project/test/test_feature_materialization.py +++ b/feathr_project/test/test_feature_materialization.py @@ -3,9 +3,10 @@ from pathlib import Path from feathr._materialization_utils import _to_materialization_config -from feathr import (BackfillTime, MaterializationSettings, FeatureQuery, +from feathr import (BackfillTime, MaterializationSettings, FeatureQuery, ObservationSettings, SparkExecutionConfiguration) from feathr import RedisSink +from feathr import OfflineSink from feathr.anchor import FeatureAnchor from feathr.dtype import BOOLEAN, FLOAT, FLOAT_VECTOR, INT32, ValueType from feathr.feature import Feature @@ -41,6 +42,32 @@ def test_feature_materialization_config(): """ assert ''.join(config.split()) == ''.join(expected_config.split()) +def test_feature_materialization_offline_config(): + backfill_time = BackfillTime(start=datetime(2020, 5, 20), end=datetime(2020, 5,20), step=timedelta(days=1)) + offlineSink = OfflineSink(output_path="abfss://feathrazuretest3fs@feathrazuretest3storage.dfs.core.windows.net/demo_data/output/hdfs_test.avro") + settings = MaterializationSettings("nycTaxiTable", + sinks=[offlineSink], + feature_names=["f_location_avg_fare", "f_location_max_fare"], + backfill_time=backfill_time) + config = _to_materialization_config(settings) + expected_config = """ + operational: { + name: nycTaxiTable + endTime: "2020-05-20 00:00:00" + endTimeFormat: "yyyy-MM-dd HH:mm:ss" + resolution: DAILY + output:[ + { + name: HDFS + params: { + path: "abfss://feathrazuretest3fs@feathrazuretest3storage.dfs.core.windows.net/demo_data/output/hdfs_test.avro" + } + } + ] + } + features: [f_location_avg_fare, f_location_max_fare] + """ + assert ''.join(config.split()) == ''.join(expected_config.split()) def test_feature_materialization_daily_schedule(): """Test back fill cutoff time for a daily range""" @@ -89,10 +116,10 @@ def test_build_feature_verbose(): anchor = FeatureAnchor(name="request_features", source=INPUT_CONTEXT, features=features) - + # Check pretty print client.build_features(anchor_list=[anchor], verbose=True) - + def test_get_offline_features_verbose(): """ Test verbose for pretty printing feature query @@ -106,7 +133,7 @@ def test_get_offline_features_verbose(): key_column_type=ValueType.INT32) feature_query = FeatureQuery(feature_list=["f_location_avg_fare"], key=location_id) - + settings = ObservationSettings( observation_path="wasbs://public@azurefeathrstorage.blob.core.windows.net/sample_data/green_tripdata_2020-04", event_timestamp_column="lpep_dropoff_datetime", @@ -114,7 +141,7 @@ def test_get_offline_features_verbose(): ) now = datetime.now() - + # set output folder based on different runtime if client.spark_runtime == 'databricks': output_path = ''.join(['dbfs:/feathrazure_cijob','_', str(now.minute), '_', str(now.second), ".parquet"]) From ca6d10ec03b3c34f0b6972f37e1ad2af0cc67930 Mon Sep 17 00:00:00 2001 From: Hangfei Lin Date: Fri, 20 May 2022 21:22:30 -0700 Subject: [PATCH 2/9] Update feature-generation.md --- docs/concepts/feature-generation.md | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/docs/concepts/feature-generation.md b/docs/concepts/feature-generation.md index d264c1b9d..adc8b61f9 100644 --- a/docs/concepts/feature-generation.md +++ b/docs/concepts/feature-generation.md @@ -5,7 +5,9 @@ parent: Feathr Concepts --- # Feature Generation -User could utilize feature generation to pre-compute and materialize pre-defined features to online and/or offline storage. This is a common practice when the feature transformation is computation intensive. +Feature generation is the process to create features from raw source data into a certain persisted storage. + +User could utilize feature generation to pre-compute and materialize pre-defined features to online and/or offline storage. This is desirable when the feature transformation is computation intensive or when the features can be reused(usually in offline setting). Feature generation is also useful in generating embedding features. Embedding distill information from large data and it is usually more compact. ## Generating Features to Online Store When we need to serve the models online, we also need to serve the features online. We provide APIs to generate features to online storage for future consumption. For example: From bc7fc29374bb8d5c497c7d6ff4c5f16ad9711ea3 Mon Sep 17 00:00:00 2001 From: Hangfei Lin Date: Fri, 20 May 2022 21:25:34 -0700 Subject: [PATCH 3/9] Update feature-generation.md --- docs/concepts/feature-generation.md | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/docs/concepts/feature-generation.md b/docs/concepts/feature-generation.md index adc8b61f9..f55a00726 100644 --- a/docs/concepts/feature-generation.md +++ b/docs/concepts/feature-generation.md @@ -54,7 +54,7 @@ After we finish running the materialization job, we can get the online features ## Generating Features to Offline Store -User could utilize feature generation to pre-compute and materialize pre-defined features to offline storage. This is a common practice when the feature transformation is computation intensive and can be re-used. For example: +This is a useful when the feature transformation is computation intensive and features can be re-used. For example, you have a feature that needs more than 24 hours to compute and the feature can be reused by more than one model training pipeline. In this case, you should consider generate features to offline. Here is an API example: ```python client = FeathrClient() offlineSink = OfflineSink(output_path="abfss://path/to/offline/hdfs_test.avro") @@ -65,4 +65,4 @@ settings = MaterializationSettings("nycTaxiMaterializationJob", client.materialize_features(settings) ``` ([MaterializationSettings API doc](https://feathr.readthedocs.io/en/latest/feathr.html#feathr.materialization_settings.MaterializationSettings), -[RedisSink API doc](https://feathr.readthedocs.io/en/latest/feathr.html#feathr.sink.RedisSink)) +[OfflineSink API doc](https://feathr.readthedocs.io/en/latest/feathr.html#feathr.sink.OfflineSink)) From 2580cf4ca83428afcee988b642d19ceb9052fc5a Mon Sep 17 00:00:00 2001 From: Hangfei Lin Date: Sat, 21 May 2022 23:43:18 -0700 Subject: [PATCH 4/9] [Feathr] Flip switch to write to storage --- feathr_project/feathr/job_utils.py | 4 ++-- feathr_project/test/test_azure_spark_e2e.py | 5 +++-- .../outputProcessor/WriteToHDFSOutputProcessor.scala | 3 +-- 3 files changed, 6 insertions(+), 6 deletions(-) diff --git a/feathr_project/feathr/job_utils.py b/feathr_project/feathr/job_utils.py index 83fe3807d..4ac1a7a88 100644 --- a/feathr_project/feathr/job_utils.py +++ b/feathr_project/feathr/job_utils.py @@ -9,10 +9,10 @@ def get_result_df(client: FeathrClient, format: str = None, res_url: str = None) -> pd.DataFrame: """Download the job result dataset from cloud as a Pandas dataframe. - + format: format override, could be "parquet", "delta", etc. res_url: output URL to download files. Note that this will not block the job so you need to make sure the job is finished and result URL contains actual data. - """ + """ res_url: str = res_url or client.get_job_result_uri(block=True, timeout_sec=1200) format: str = format or client.get_job_tags().get(OUTPUT_FORMAT, "") tmp_dir = tempfile.TemporaryDirectory() diff --git a/feathr_project/test/test_azure_spark_e2e.py b/feathr_project/test/test_azure_spark_e2e.py index f3c9cdf98..5cededff4 100644 --- a/feathr_project/test/test_azure_spark_e2e.py +++ b/feathr_project/test/test_azure_spark_e2e.py @@ -32,7 +32,7 @@ def test_feathr_materialize_to_offline(): backfill_time = BackfillTime(start=datetime( 2020, 5, 20), end=datetime(2020, 5, 20), step=timedelta(days=1)) - offline_sink = OfflineSink(output_path="abfss://feathrazuretest3fs@feathrazuretest3storage.dfs.core.windows.net/demo_data/output/hdfs_test.avro") + offline_sink = OfflineSink(output_path="abfss://feathrazuretest3fs@feathrazuretest3storage.dfs.core.windows.net/materialize_offline_test_data/") settings = MaterializationSettings("nycTaxiTable", sinks=[offline_sink], feature_names=[ @@ -43,7 +43,8 @@ def test_feathr_materialize_to_offline(): client.wait_job_to_finish(timeout_sec=900) # download result and just assert the returned result is not empty - res_df = get_result_df(client) + # by default, it will write to a folder appended with date + res_df = get_result_df(client, "avro", "abfss://feathrazuretest3fs@feathrazuretest3storage.dfs.core.windows.net/materialize_offline_test_data/df0/daily/2020/05/20") assert res_df.shape[0] > 0 def test_feathr_online_store_agg_features(): diff --git a/src/main/scala/com/linkedin/feathr/offline/generation/outputProcessor/WriteToHDFSOutputProcessor.scala b/src/main/scala/com/linkedin/feathr/offline/generation/outputProcessor/WriteToHDFSOutputProcessor.scala index 64a16aee3..d01deabd7 100644 --- a/src/main/scala/com/linkedin/feathr/offline/generation/outputProcessor/WriteToHDFSOutputProcessor.scala +++ b/src/main/scala/com/linkedin/feathr/offline/generation/outputProcessor/WriteToHDFSOutputProcessor.scala @@ -145,8 +145,7 @@ private[offline] class WriteToHDFSOutputProcessor(val config: OutputProcessorCon } val featuresToDF = taggedFeatureNames.map(featureToDF => (featureToDF, (augmentedDF, header))).toMap - // Note that this function returns the resulting output w/o writing data file system. - FeatureDataHDFSProcessUtils.processFeatureDataHDFS(ss, featuresToDF, parentPath, config, skipWrite = true, endTimeOpt, timestampOpt) + FeatureDataHDFSProcessUtils.processFeatureDataHDFS(ss, featuresToDF, parentPath, config, skipWrite = false, endTimeOpt, timestampOpt) } // path parameter name From 25e04f84ff7232245c5acfeed9709b1ecbb3e902 Mon Sep 17 00:00:00 2001 From: Hangfei Lin Date: Sat, 21 May 2022 23:54:47 -0700 Subject: [PATCH 5/9] [feahtr] Update docs --- docs/concepts/feature-generation.md | 17 ++++++++++++++++- feathr_project/test/test_azure_spark_e2e.py | 4 ++-- .../test/test_user_workspace/feathr_config.yaml | 2 +- 3 files changed, 19 insertions(+), 4 deletions(-) diff --git a/docs/concepts/feature-generation.md b/docs/concepts/feature-generation.md index f55a00726..b67637880 100644 --- a/docs/concepts/feature-generation.md +++ b/docs/concepts/feature-generation.md @@ -57,12 +57,27 @@ After we finish running the materialization job, we can get the online features This is a useful when the feature transformation is computation intensive and features can be re-used. For example, you have a feature that needs more than 24 hours to compute and the feature can be reused by more than one model training pipeline. In this case, you should consider generate features to offline. Here is an API example: ```python client = FeathrClient() -offlineSink = OfflineSink(output_path="abfss://path/to/offline/hdfs_test.avro") +offlineSink = OfflineSink(output_path="abfss://feathrazuretest3fs@feathrazuretest3storage.dfs.core.windows.net/materialize_offline_test_data/") # Materialize two features into a Offline store. settings = MaterializationSettings("nycTaxiMaterializationJob", sinks=[offlineSink], feature_names=["f_location_avg_fare", "f_location_max_fare"]) client.materialize_features(settings) ``` +This will generate features on latest date(assuming it's `2022/05/21`) and output data to the following path: `abfss://feathrazuretest3fs@feathrazuretest3storage.dfs.core.windows.net/materialize_offline_test_data/df0/daily/2022/05/21` + + +You can also specify a BackfillTime so the features will be generated for those dates. For example: +```Python +backfill_time = BackfillTime(start=datetime( + 2020, 5, 20), end=datetime(2020, 5, 20), step=timedelta(days=1)) +offline_sink = OfflineSink(output_path="abfss://feathrazuretest3fs@feathrazuretest3storage.dfs.core.windows.net/materialize_offline_test_data/") +settings = MaterializationSettings("nycTaxiTable", + sinks=[offline_sink], + feature_names=[ + "f_location_avg_fare", "f_location_max_fare"], + backfill_time=backfill_time) +``` +This will generate features only for 2020/05/20 for me and it will be in folder: `abfss://feathrazuretest3fs@feathrazuretest3storage.dfs.core.windows.net/materialize_offline_test_data/df0/daily/2020/05/20` ([MaterializationSettings API doc](https://feathr.readthedocs.io/en/latest/feathr.html#feathr.materialization_settings.MaterializationSettings), [OfflineSink API doc](https://feathr.readthedocs.io/en/latest/feathr.html#feathr.sink.OfflineSink)) diff --git a/feathr_project/test/test_azure_spark_e2e.py b/feathr_project/test/test_azure_spark_e2e.py index 5cededff4..bcafeb4e4 100644 --- a/feathr_project/test/test_azure_spark_e2e.py +++ b/feathr_project/test/test_azure_spark_e2e.py @@ -36,8 +36,8 @@ def test_feathr_materialize_to_offline(): settings = MaterializationSettings("nycTaxiTable", sinks=[offline_sink], feature_names=[ - "f_location_avg_fare", "f_location_max_fare"], - backfill_time=backfill_time) + "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=900) diff --git a/feathr_project/test/test_user_workspace/feathr_config.yaml b/feathr_project/test/test_user_workspace/feathr_config.yaml index ab9c02432..e80c12bd4 100644 --- a/feathr_project/test/test_user_workspace/feathr_config.yaml +++ b/feathr_project/test/test_user_workspace/feathr_config.yaml @@ -67,7 +67,7 @@ offline_store: spark_config: # choice for spark runtime. Currently support: azure_synapse, databricks # The `databricks` configs will be ignored if `azure_synapse` is set and vice versa. - spark_cluster: 'databricks' + spark_cluster: 'azure_synapse' # configure number of parts for the spark output for feature generation job spark_result_output_parts: '1' From 359a7e49dd13950fc66532a98792cec6e1076578 Mon Sep 17 00:00:00 2001 From: Hangfei Lin Date: Sun, 22 May 2022 10:11:07 -0700 Subject: [PATCH 6/9] Fix databrics test --- feathr_project/test/test_azure_spark_e2e.py | 14 ++++++++++---- .../test/test_user_workspace/feathr_config.yaml | 2 +- .../WriteToHDFSOutputProcessor.scala | 4 +++- 3 files changed, 14 insertions(+), 6 deletions(-) diff --git a/feathr_project/test/test_azure_spark_e2e.py b/feathr_project/test/test_azure_spark_e2e.py index bcafeb4e4..45481a215 100644 --- a/feathr_project/test/test_azure_spark_e2e.py +++ b/feathr_project/test/test_azure_spark_e2e.py @@ -32,19 +32,25 @@ def test_feathr_materialize_to_offline(): backfill_time = BackfillTime(start=datetime( 2020, 5, 20), end=datetime(2020, 5, 20), step=timedelta(days=1)) - offline_sink = OfflineSink(output_path="abfss://feathrazuretest3fs@feathrazuretest3storage.dfs.core.windows.net/materialize_offline_test_data/") + + 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 = OfflineSink(output_path=output_path) settings = MaterializationSettings("nycTaxiTable", sinks=[offline_sink], feature_names=[ - "f_location_avg_fare", "f_location_max_fare"]) - # backfill_time=backfill_time) + "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=900) # 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", "abfss://feathrazuretest3fs@feathrazuretest3storage.dfs.core.windows.net/materialize_offline_test_data/df0/daily/2020/05/20") + 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(): diff --git a/feathr_project/test/test_user_workspace/feathr_config.yaml b/feathr_project/test/test_user_workspace/feathr_config.yaml index e80c12bd4..ab9c02432 100644 --- a/feathr_project/test/test_user_workspace/feathr_config.yaml +++ b/feathr_project/test/test_user_workspace/feathr_config.yaml @@ -67,7 +67,7 @@ offline_store: spark_config: # choice for spark runtime. Currently support: azure_synapse, databricks # The `databricks` configs will be ignored if `azure_synapse` is set and vice versa. - spark_cluster: 'azure_synapse' + spark_cluster: 'databricks' # configure number of parts for the spark output for feature generation job spark_result_output_parts: '1' diff --git a/src/main/scala/com/linkedin/feathr/offline/generation/outputProcessor/WriteToHDFSOutputProcessor.scala b/src/main/scala/com/linkedin/feathr/offline/generation/outputProcessor/WriteToHDFSOutputProcessor.scala index d01deabd7..740256686 100644 --- a/src/main/scala/com/linkedin/feathr/offline/generation/outputProcessor/WriteToHDFSOutputProcessor.scala +++ b/src/main/scala/com/linkedin/feathr/offline/generation/outputProcessor/WriteToHDFSOutputProcessor.scala @@ -145,7 +145,9 @@ private[offline] class WriteToHDFSOutputProcessor(val config: OutputProcessorCon } val featuresToDF = taggedFeatureNames.map(featureToDF => (featureToDF, (augmentedDF, header))).toMap - FeatureDataHDFSProcessUtils.processFeatureDataHDFS(ss, featuresToDF, parentPath, config, skipWrite = false, endTimeOpt, timestampOpt) + // If it's local, we can't write to HDFS. + val skipWrite = if (ss.sparkContext.isLocal) true else false + FeatureDataHDFSProcessUtils.processFeatureDataHDFS(ss, featuresToDF, parentPath, config, skipWrite = skipWrite, endTimeOpt, timestampOpt) } // path parameter name From cd9b9dc348a3f2c599acab6a6ea6b20bbcdf6537 Mon Sep 17 00:00:00 2001 From: Hangfei Lin Date: Sun, 22 May 2022 10:17:46 -0700 Subject: [PATCH 7/9] improve docs --- feathr_project/feathr/sink.py | 37 ++++++++++++++++++----------------- 1 file changed, 19 insertions(+), 18 deletions(-) diff --git a/feathr_project/feathr/sink.py b/feathr_project/feathr/sink.py index d4967755c..c0fe7b256 100644 --- a/feathr_project/feathr/sink.py +++ b/feathr_project/feathr/sink.py @@ -42,31 +42,32 @@ def to_feature_config(self) -> str: return msg -# Sample generated HOCON config: -# operational: { -# name: testFeatureGen -# endTime: 2019-05-01 -# endTimeFormat: "yyyy-MM-dd" -# resolution: DAILY -# output:[ -# { -# name: HDFS -# params: { -# path: "/user/featureGen/hdfsResult/" -# } -# } -# ] -# } -# features: [mockdata_a_ct_gen, mockdata_a_sample_gen] class OfflineSink(Sink): - """Offline Hadoop-compatible sink use to store offline feature data, can be used in batch job. + """Offline Hadoop HDFS-compatible(HDFS, delta lake, Azure blog storage etc) sink that is used to store feature data. + The result is in AVRO format. Attributes: - table_name: output table name + output_path: output path """ def __init__(self, output_path: str) -> None: self.output_path = output_path + # Sample generated HOCON config: + # operational: { + # name: testFeatureGen + # endTime: 2019-05-01 + # endTimeFormat: "yyyy-MM-dd" + # resolution: DAILY + # output:[ + # { + # name: HDFS + # params: { + # path: "/user/featureGen/hdfsResult/" + # } + # } + # ] + # } + # features: [mockdata_a_ct_gen, mockdata_a_sample_gen] def to_feature_config(self) -> str: """Produce the config used in feature materialization""" tm = Template(""" From 4369a59ff8742b4bddb320348d71f05333c3d9b2 Mon Sep 17 00:00:00 2001 From: Hangfei Lin Date: Mon, 23 May 2022 20:43:33 -0700 Subject: [PATCH 8/9] fix markdown formatting --- docs/concepts/feature-generation.md | 15 +++++++++++---- 1 file changed, 11 insertions(+), 4 deletions(-) diff --git a/docs/concepts/feature-generation.md b/docs/concepts/feature-generation.md index b67637880..69f9f400e 100644 --- a/docs/concepts/feature-generation.md +++ b/docs/concepts/feature-generation.md @@ -50,11 +50,15 @@ res = client.get_online_features('nycTaxiDemoFeature', '265', [ ``` ([client.get_online_features API doc](https://feathr.readthedocs.io/en/latest/feathr.html#feathr.client.FeathrClient.get_online_features)) -After we finish running the materialization job, we can get the online features by querying the feature name, with the corresponding keys. In the example above, we query the online features called `f_location_avg_fare` and `f_location_max_fare`, and query with a key `265` (which is the location ID). +After we finish running the materialization job, we can get the online features by querying the feature name, with the +corresponding keys. In the example above, we query the online features called `f_location_avg_fare` and +`f_location_max_fare`, and query with a key `265` (which is the location ID). ## Generating Features to Offline Store -This is a useful when the feature transformation is computation intensive and features can be re-used. For example, you have a feature that needs more than 24 hours to compute and the feature can be reused by more than one model training pipeline. In this case, you should consider generate features to offline. Here is an API example: +This is a useful when the feature transformation is computation intensive and features can be re-used. For example, you +have a feature that needs more than 24 hours to compute and the feature can be reused by more than one model training +pipeline. In this case, you should consider generate features to offline. Here is an API example: ```python client = FeathrClient() offlineSink = OfflineSink(output_path="abfss://feathrazuretest3fs@feathrazuretest3storage.dfs.core.windows.net/materialize_offline_test_data/") @@ -64,7 +68,8 @@ settings = MaterializationSettings("nycTaxiMaterializationJob", feature_names=["f_location_avg_fare", "f_location_max_fare"]) client.materialize_features(settings) ``` -This will generate features on latest date(assuming it's `2022/05/21`) and output data to the following path: `abfss://feathrazuretest3fs@feathrazuretest3storage.dfs.core.windows.net/materialize_offline_test_data/df0/daily/2022/05/21` +This will generate features on latest date(assuming it's `2022/05/21`) and output data to the following path: +`abfss://feathrazuretest3fs@feathrazuretest3storage.dfs.core.windows.net/materialize_offline_test_data/df0/daily/2022/05/21` You can also specify a BackfillTime so the features will be generated for those dates. For example: @@ -78,6 +83,8 @@ settings = MaterializationSettings("nycTaxiTable", "f_location_avg_fare", "f_location_max_fare"], backfill_time=backfill_time) ``` -This will generate features only for 2020/05/20 for me and it will be in folder: `abfss://feathrazuretest3fs@feathrazuretest3storage.dfs.core.windows.net/materialize_offline_test_data/df0/daily/2020/05/20` +This will generate features only for 2020/05/20 for me and it will be in folder: +`abfss://feathrazuretest3fs@feathrazuretest3storage.dfs.core.windows.net/materialize_offline_test_data/df0/daily/2020/05/20` + ([MaterializationSettings API doc](https://feathr.readthedocs.io/en/latest/feathr.html#feathr.materialization_settings.MaterializationSettings), [OfflineSink API doc](https://feathr.readthedocs.io/en/latest/feathr.html#feathr.sink.OfflineSink)) From 3650b6034f20aa1f7c26bab08bb7245fb21760e3 Mon Sep 17 00:00:00 2001 From: Hangfei Lin Date: Mon, 23 May 2022 20:47:24 -0700 Subject: [PATCH 9/9] Rename to HdfsSink --- docs/concepts/feature-generation.md | 6 +++--- feathr_project/feathr/__init__.py | 2 +- feathr_project/feathr/sink.py | 2 +- feathr_project/test/test_azure_spark_e2e.py | 6 +++--- feathr_project/test/test_feature_materialization.py | 4 ++-- 5 files changed, 10 insertions(+), 10 deletions(-) diff --git a/docs/concepts/feature-generation.md b/docs/concepts/feature-generation.md index 69f9f400e..a04ff1719 100644 --- a/docs/concepts/feature-generation.md +++ b/docs/concepts/feature-generation.md @@ -61,7 +61,7 @@ have a feature that needs more than 24 hours to compute and the feature can be r pipeline. In this case, you should consider generate features to offline. Here is an API example: ```python client = FeathrClient() -offlineSink = OfflineSink(output_path="abfss://feathrazuretest3fs@feathrazuretest3storage.dfs.core.windows.net/materialize_offline_test_data/") +offlineSink = HdfsSink(output_path="abfss://feathrazuretest3fs@feathrazuretest3storage.dfs.core.windows.net/materialize_offline_test_data/") # Materialize two features into a Offline store. settings = MaterializationSettings("nycTaxiMaterializationJob", sinks=[offlineSink], @@ -76,7 +76,7 @@ You can also specify a BackfillTime so the features will be generated for those ```Python backfill_time = BackfillTime(start=datetime( 2020, 5, 20), end=datetime(2020, 5, 20), step=timedelta(days=1)) -offline_sink = OfflineSink(output_path="abfss://feathrazuretest3fs@feathrazuretest3storage.dfs.core.windows.net/materialize_offline_test_data/") +offline_sink = HdfsSink(output_path="abfss://feathrazuretest3fs@feathrazuretest3storage.dfs.core.windows.net/materialize_offline_test_data/") settings = MaterializationSettings("nycTaxiTable", sinks=[offline_sink], feature_names=[ @@ -87,4 +87,4 @@ This will generate features only for 2020/05/20 for me and it will be in folder: `abfss://feathrazuretest3fs@feathrazuretest3storage.dfs.core.windows.net/materialize_offline_test_data/df0/daily/2020/05/20` ([MaterializationSettings API doc](https://feathr.readthedocs.io/en/latest/feathr.html#feathr.materialization_settings.MaterializationSettings), -[OfflineSink API doc](https://feathr.readthedocs.io/en/latest/feathr.html#feathr.sink.OfflineSink)) +[HdfsSink API doc](https://feathr.readthedocs.io/en/latest/feathr.html#feathr.sink.HdfsSink)) diff --git a/feathr_project/feathr/__init__.py b/feathr_project/feathr/__init__.py index 6957692ae..433c4aaae 100644 --- a/feathr_project/feathr/__init__.py +++ b/feathr_project/feathr/__init__.py @@ -8,7 +8,7 @@ from .transformation import * from .typed_key import * from .materialization_settings import (BackfillTime, MaterializationSettings) -from .sink import RedisSink, OfflineSink +from .sink import RedisSink, HdfsSink from .query_feature_list import FeatureQuery from .lookup_feature import LookupFeature from .aggregation import Aggregation diff --git a/feathr_project/feathr/sink.py b/feathr_project/feathr/sink.py index c0fe7b256..9d3f3d56e 100644 --- a/feathr_project/feathr/sink.py +++ b/feathr_project/feathr/sink.py @@ -42,7 +42,7 @@ def to_feature_config(self) -> str: return msg -class OfflineSink(Sink): +class HdfsSink(Sink): """Offline Hadoop HDFS-compatible(HDFS, delta lake, Azure blog storage etc) sink that is used to store feature data. The result is in AVRO format. diff --git a/feathr_project/test/test_azure_spark_e2e.py b/feathr_project/test/test_azure_spark_e2e.py index 45481a215..4f2dcd717 100644 --- a/feathr_project/test/test_azure_spark_e2e.py +++ b/feathr_project/test/test_azure_spark_e2e.py @@ -10,7 +10,7 @@ from feathr import (BackfillTime, MaterializationSettings) from feathr import FeatureQuery from feathr import ObservationSettings -from feathr import RedisSink, OfflineSink +from feathr import RedisSink, HdfsSink from feathr import TypedKey from feathrcli.cli import init import pytest @@ -20,7 +20,7 @@ # the feature configs are from feathr_project/data/feathr_user_workspace def test_feathr_materialize_to_offline(): """ - Test FeathrClient() OfflineSink. + Test FeathrClient() HdfsSink. """ online_test_table = get_online_test_table_name("nycTaxiCITable") @@ -38,7 +38,7 @@ def test_feathr_materialize_to_offline(): 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 = OfflineSink(output_path=output_path) + offline_sink = HdfsSink(output_path=output_path) settings = MaterializationSettings("nycTaxiTable", sinks=[offline_sink], feature_names=[ diff --git a/feathr_project/test/test_feature_materialization.py b/feathr_project/test/test_feature_materialization.py index c95ef2497..72aed5792 100644 --- a/feathr_project/test/test_feature_materialization.py +++ b/feathr_project/test/test_feature_materialization.py @@ -6,7 +6,7 @@ from feathr import (BackfillTime, MaterializationSettings, FeatureQuery, ObservationSettings, SparkExecutionConfiguration) from feathr import RedisSink -from feathr import OfflineSink +from feathr import HdfsSink from feathr.anchor import FeatureAnchor from feathr.dtype import BOOLEAN, FLOAT, FLOAT_VECTOR, INT32, ValueType from feathr.feature import Feature @@ -44,7 +44,7 @@ def test_feature_materialization_config(): def test_feature_materialization_offline_config(): backfill_time = BackfillTime(start=datetime(2020, 5, 20), end=datetime(2020, 5,20), step=timedelta(days=1)) - offlineSink = OfflineSink(output_path="abfss://feathrazuretest3fs@feathrazuretest3storage.dfs.core.windows.net/demo_data/output/hdfs_test.avro") + offlineSink = HdfsSink(output_path="abfss://feathrazuretest3fs@feathrazuretest3storage.dfs.core.windows.net/demo_data/output/hdfs_test.avro") settings = MaterializationSettings("nycTaxiTable", sinks=[offlineSink], feature_names=["f_location_avg_fare", "f_location_max_fare"],