diff --git a/docs/concepts/feature-generation.md b/docs/concepts/feature-generation.md index 7e18557fe..a04ff1719 100644 --- a/docs/concepts/feature-generation.md +++ b/docs/concepts/feature-generation.md @@ -5,10 +5,12 @@ parent: Feathr Concepts --- # Feature Generation +Feature generation is the process to create features from raw source data into a certain persisted storage. -## 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 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. -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: +## 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: ```python client = FeathrClient() redisSink = RedisSink(table_name="nycTaxiDemoFeature") @@ -48,4 +50,41 @@ 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: +```python +client = FeathrClient() +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], + 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 = HdfsSink(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), +[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 7e74ef90d..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 +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/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/feathr/sink.py b/feathr_project/feathr/sink.py index 75b5b7750..9d3f3d56e 100644 --- a/feathr_project/feathr/sink.py +++ b/feathr_project/feathr/sink.py @@ -39,4 +39,44 @@ def to_feature_config(self) -> str: } """) msg = tm.render(source=self) - return msg \ No newline at end of file + return msg + + +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. + + Attributes: + 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(""" + { + 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..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 +from feathr import RedisSink, HdfsSink from feathr import TypedKey from feathrcli.cli import init import pytest @@ -18,6 +18,41 @@ 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() HdfsSink. + """ + + 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)) + + 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=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", 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. @@ -69,7 +104,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..72aed5792 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 HdfsSink 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 = 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"], + 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"]) 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..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,8 +145,9 @@ 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) + // 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