feathr-ai · hangfei · May 25, 2022 · May 21, 2022 · May 21, 2022 · May 21, 2022
diff --git 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
@@ -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
@@ -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
@@ -39,4 +39,44 @@ def to_feature_config(self) -> str:
             }
         """)
         msg = tm.render(source=self)
-        return msg
+        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
diff --git 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
@@ -10,14 +10,49 @@
 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
 
 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
@@ -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,15 +133,15 @@ 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",
         timestamp_format="yyyy-MM-dd HH:mm:ss"
     )
 
     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/...a/com/linkedin/feathr/offline/generation/outputProcessor/WriteToHDFSOutputProcessor.scala b/...a/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