Skip to content

Commit 799fac0

Browse files
xiaoyongzhuxiaoyongzhu
authored
Decouple build feature code (#838)
* Restructure code to encapsulate the `save_to_feature_config_from_context` * Update client.py * fix merge issues * Update config_helper.py * fix comments
1 parent c21d89d commit 799fac0

5 files changed

Lines changed: 231 additions & 408 deletions

File tree

‎feathr_project/feathr/client.py‎

Lines changed: 36 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -23,7 +23,6 @@
2323
from feathr.definition.settings import ObservationSettings
2424
from feathr.definition.sink import Sink
2525
from feathr.protobuf.featureValue_pb2 import FeatureValue
26-
from feathr.registry.feature_registry import default_registry_client
2726
from feathr.spark_provider._databricks_submission import _FeathrDatabricksJobLauncher
2827
from feathr.spark_provider._localspark_submission import _FeathrLocalSparkJobLauncher
2928
from feathr.spark_provider._synapse_submission import _FeathrSynapseJobLauncher
@@ -34,8 +33,14 @@
3433
from feathr.utils.feature_printer import FeaturePrinter
3534
from feathr.utils.spark_job_params import FeatureGenerationJobParams, FeatureJoinJobParams
3635
from feathr.definition.source import InputContext
36+
from azure.identity import DefaultAzureCredential
37+
from jinja2 import Template
38+
from loguru import logger
39+
from feathr.definition.config_helper import FeathrConfigHelper
40+
from pyhocon import ConfigFactory
41+
from feathr.registry._feathr_registry_client import _FeatureRegistry
42+
from feathr.registry._feature_registry_purview import _PurviewRegistry
3743
from feathr.version import get_version
38-
3944
class FeathrClient(object):
4045
"""Feathr client.
4146
@@ -170,10 +175,24 @@ def __init__(self, config_path:str = "./feathr_config.yaml", local_workspace_dir
170175

171176
self.secret_names = []
172177

173-
# initialize registry
174-
self.registry = default_registry_client(self.project_name, config_path=config_path, credential=self.credential)
178+
# initialize config helper
179+
self.config_helper = FeathrConfigHelper()
175180

176-
logger.info(f"Feathr Client {get_version()} initialized successfully")
181+
# initialize registry
182+
self.registry = None
183+
registry_endpoint = self.envutils.get_environment_variable_with_default("feature_registry", "api_endpoint")
184+
azure_purview_name = self.envutils.get_environment_variable_with_default('feature_registry', 'purview', 'purview_name')
185+
if registry_endpoint:
186+
self.registry = _FeatureRegistry(self.project_name, endpoint=registry_endpoint, project_tags=project_registry_tag, credential=credential)
187+
elif azure_purview_name:
188+
registry_delimiter = self.envutils.get_environment_variable_with_default('feature_registry', 'purview', 'delimiter')
189+
# initialize the registry no matter whether we set purview name or not, given some of the methods are used there.
190+
self.registry = _PurviewRegistry(self.project_name, azure_purview_name, registry_delimiter, project_registry_tag, config_path = config_path, credential=credential)
191+
else:
192+
# no registry configured
193+
logger.info("Feathr registry is not configured. Consider setting the Feathr registry component for richer feature store experience.")
194+
195+
logger.info(f"Feathr client {get_version()} initialized successfully.")
177196

178197
def _check_required_environment_variables_exist(self):
179198
"""Checks if the required environment variables(form feathr_config.yaml) is set.
@@ -197,7 +216,7 @@ def register_features(self, from_context: bool = True):
197216
if from_context:
198217
# make sure those items are in `self`
199218
if 'anchor_list' in dir(self) and 'derived_feature_list' in dir(self):
200-
self.registry.save_to_feature_config_from_context(self.anchor_list, self.derived_feature_list, self.local_workspace_dir)
219+
self.config_helper.save_to_feature_config_from_context(self.anchor_list, self.derived_feature_list, self.local_workspace_dir)
201220
self.registry.register_features(self.local_workspace_dir, from_context=from_context, anchor_list=self.anchor_list, derived_feature_list=self.derived_feature_list)
202221
else:
203222
raise RuntimeError("Please call FeathrClient.build_features() first in order to register features")
@@ -224,9 +243,8 @@ def build_features(self, anchor_list: List[FeatureAnchor] = [], derived_feature_
224243
else:
225244
source_names[anchor.source.name] = anchor.source
226245

227-
preprocessingPyudfManager = _PreprocessingPyudfManager()
228246
_PreprocessingPyudfManager.build_anchor_preprocessing_metadata(anchor_list, self.local_workspace_dir)
229-
self.registry.save_to_feature_config_from_context(anchor_list, derived_feature_list, self.local_workspace_dir)
247+
self.config_helper.save_to_feature_config_from_context(anchor_list, derived_feature_list, self.local_workspace_dir)
230248
self.anchor_list = anchor_list
231249
self.derived_feature_list = derived_feature_list
232250

@@ -470,7 +488,7 @@ def get_offline_features(self,
470488
# otherwise users will be confused on what are the available features
471489
# in build_features it will assign anchor_list and derived_feature_list variable, hence we are checking if those two variables exist to make sure the above condition is met
472490
if 'anchor_list' in dir(self) and 'derived_feature_list' in dir(self):
473-
self.registry.save_to_feature_config_from_context(self.anchor_list, self.derived_feature_list, self.local_workspace_dir)
491+
self.config_helper.save_to_feature_config_from_context(self.anchor_list, self.derived_feature_list, self.local_workspace_dir)
474492
else:
475493
raise RuntimeError("Please call FeathrClient.build_features() first in order to get offline features")
476494

@@ -678,7 +696,7 @@ def materialize_features(self, settings: MaterializationSettings, execution_conf
678696
# otherwise users will be confused on what are the available features
679697
# in build_features it will assign anchor_list and derived_feature_list variable, hence we are checking if those two variables exist to make sure the above condition is met
680698
if 'anchor_list' in dir(self) and 'derived_feature_list' in dir(self):
681-
self.registry.save_to_feature_config_from_context(self.anchor_list, self.derived_feature_list, self.local_workspace_dir)
699+
self.config_helper.save_to_feature_config_from_context(self.anchor_list, self.derived_feature_list, self.local_workspace_dir)
682700
else:
683701
raise RuntimeError("Please call FeathrClient.build_features() first in order to materialize the features")
684702

@@ -772,7 +790,7 @@ def _get_s3_config_str(self):
772790
# keys can't be only accessed through environment
773791
access_key = self.envutils.get_environment_variable('S3_ACCESS_KEY')
774792
secret_key = self.envutils.get_environment_variable('S3_SECRET_KEY')
775-
# HOCCON format will be parsed by the Feathr job
793+
# HOCON format will be parsed by the Feathr job
776794
config_str = """
777795
S3_ENDPOINT: {S3_ENDPOINT}
778796
S3_ACCESS_KEY: "{S3_ACCESS_KEY}"
@@ -787,7 +805,7 @@ def _get_adls_config_str(self):
787805
# if ADLS Account is set in the feathr_config, then we need other environment variables
788806
# keys can't be only accessed through environment
789807
key = self.envutils.get_environment_variable('ADLS_KEY')
790-
# HOCCON format will be parsed by the Feathr job
808+
# HOCON format will be parsed by the Feathr job
791809
config_str = """
792810
ADLS_ACCOUNT: {ADLS_ACCOUNT}
793811
ADLS_KEY: "{ADLS_KEY}"
@@ -801,7 +819,7 @@ def _get_blob_config_str(self):
801819
# if BLOB Account is set in the feathr_config, then we need other environment variables
802820
# keys can't be only accessed through environment
803821
key = self.envutils.get_environment_variable('BLOB_KEY')
804-
# HOCCON format will be parsed by the Feathr job
822+
# HOCON format will be parsed by the Feathr job
805823
config_str = """
806824
BLOB_ACCOUNT: {BLOB_ACCOUNT}
807825
BLOB_KEY: "{BLOB_KEY}"
@@ -817,7 +835,7 @@ def _get_sql_config_str(self):
817835
driver = self.envutils.get_environment_variable('JDBC_DRIVER')
818836
auth_flag = self.envutils.get_environment_variable('JDBC_AUTH_FLAG')
819837
token = self.envutils.get_environment_variable('JDBC_TOKEN')
820-
# HOCCON format will be parsed by the Feathr job
838+
# HOCON format will be parsed by the Feathr job
821839
config_str = """
822840
JDBC_TABLE: {JDBC_TABLE}
823841
JDBC_USER: {JDBC_USER}
@@ -834,7 +852,7 @@ def _get_monitoring_config_str(self):
834852
user = self.envutils.get_environment_variable_with_default('monitoring', 'database', 'sql', 'user')
835853
password = self.envutils.get_environment_variable('MONITORING_DATABASE_SQL_PASSWORD')
836854
if url:
837-
# HOCCON format will be parsed by the Feathr job
855+
# HOCON format will be parsed by the Feathr job
838856
config_str = """
839857
MONITORING_DATABASE_SQL_URL: "{url}"
840858
MONITORING_DATABASE_SQL_USER: {user}
@@ -852,7 +870,7 @@ def _get_snowflake_config_str(self):
852870
sf_role = self.envutils.get_environment_variable_with_default('offline_store', 'snowflake', 'role')
853871
sf_warehouse = self.envutils.get_environment_variable_with_default('offline_store', 'snowflake', 'warehouse')
854872
sf_password = self.envutils.get_environment_variable('JDBC_SF_PASSWORD')
855-
# HOCCON format will be parsed by the Feathr job
873+
# HOCON format will be parsed by the Feathr job
856874
config_str = """
857875
JDBC_SF_URL: {JDBC_SF_URL}
858876
JDBC_SF_USER: {JDBC_SF_USER}
@@ -866,7 +884,7 @@ def _get_kafka_config_str(self):
866884
"""Construct the Kafka config string. The endpoint, access key, secret key, and other parameters can be set via
867885
environment variables."""
868886
sasl = self.envutils.get_environment_variable('KAFKA_SASL_JAAS_CONFIG')
869-
# HOCCON format will be parsed by the Feathr job
887+
# HOCON format will be parsed by the Feathr job
870888
config_str = """
871889
KAFKA_SASL_JAAS_CONFIG: "{sasl}"
872890
""".format(sasl=sasl)
@@ -899,4 +917,4 @@ def _reshape_config_str(self, config_str:str):
899917
if self.spark_runtime == 'local':
900918
return "'{" + config_str + "}'"
901919
else:
902-
return config_str
920+
return config_str

‎feathr_project/feathr/definition/config_helper.py‎

Lines changed: 193 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,193 @@
1+
from feathr.definition.dtype import *
2+
from feathr.registry.registry_utils import *
3+
from feathr.utils._file_utils import write_to_file
4+
from feathr.definition.anchor import FeatureAnchor
5+
from feathr.constants import *
6+
from feathr.definition.feature import Feature, FeatureType,FeatureBase
7+
from feathr.definition.feature_derivations import DerivedFeature
8+
from feathr.definition.repo_definitions import RepoDefinitions
9+
from feathr.definition.source import HdfsSource, InputContext, JdbcSource, Source
10+
from feathr.definition.transformation import (ExpressionTransformation, Transformation,
11+
WindowAggTransformation)
12+
from feathr.definition.typed_key import TypedKey
13+
from feathr.registry.feature_registry import FeathrRegistry
14+
from feathr.definition.repo_definitions import RepoDefinitions
15+
from pathlib import Path
16+
from jinja2 import Template
17+
import sys
18+
from feathr.utils._file_utils import write_to_file
19+
import importlib
20+
import os
21+
22+
class FeathrConfigHelper(object):
23+
def __init__(self) -> None:
24+
pass
25+
def _get_py_files(self, path: Path) -> List[Path]:
26+
"""Get all Python files under path recursively, excluding __init__.py"""
27+
py_files = []
28+
for item in path.glob('**/*.py'):
29+
if "__init__.py" != item.name:
30+
py_files.append(item)
31+
return py_files
32+
33+
def _convert_to_module_path(self, path: Path, workspace_path: Path) -> str:
34+
"""Convert a Python file path to its module path so that we can import it later"""
35+
prefix = os.path.commonprefix(
36+
[path.resolve(), workspace_path.resolve()])
37+
resolved_path = str(path.resolve())
38+
module_path = resolved_path[len(prefix): -len(".py")]
39+
# Convert features under nested folder to module name
40+
# e.g. /path/to/pyfile will become path.to.pyfile
41+
return (
42+
module_path
43+
.lstrip('/')
44+
.replace("/", ".")
45+
)
46+
47+
def _extract_features_from_context(self, anchor_list, derived_feature_list, result_path: Path) -> RepoDefinitions:
48+
"""Collect feature definitions from the context instead of python files"""
49+
definitions = RepoDefinitions(
50+
sources=set(),
51+
features=set(),
52+
transformations=set(),
53+
feature_anchors=set(),
54+
derived_features=set()
55+
)
56+
for derived_feature in derived_feature_list:
57+
if isinstance(derived_feature, DerivedFeature):
58+
definitions.derived_features.add(derived_feature)
59+
definitions.transformations.add(
60+
vars(derived_feature)["transform"])
61+
else:
62+
raise RuntimeError(f"Please make sure you pass a list of `DerivedFeature` objects to the `derived_feature_list` argument. {str(type(derived_feature))} is detected.")
63+
64+
for anchor in anchor_list:
65+
# obj is `FeatureAnchor`
66+
definitions.feature_anchors.add(anchor)
67+
# add the source section of this `FeatureAnchor` object
68+
definitions.sources.add(vars(anchor)['source'])
69+
for feature in vars(anchor)['features']:
70+
# get the transformation object from `Feature` or `DerivedFeature`
71+
if isinstance(feature, Feature):
72+
# feature is of type `Feature`
73+
definitions.features.add(feature)
74+
definitions.transformations.add(vars(feature)["transform"])
75+
else:
76+
77+
raise RuntimeError(f"Please make sure you pass a list of `Feature` objects. {str(type(feature))} is detected.")
78+
79+
return definitions
80+
81+
def _extract_features(self, workspace_path: Path) -> RepoDefinitions:
82+
"""Collect feature definitions from the python file, convert them into feature config and save them locally"""
83+
os.chdir(workspace_path)
84+
# Add workspace path to system path so that we can load features defined in Python via import_module
85+
sys.path.append(str(workspace_path))
86+
definitions = RepoDefinitions(
87+
sources=set(),
88+
features=set(),
89+
transformations=set(),
90+
feature_anchors=set(),
91+
derived_features=set()
92+
)
93+
for py_file in self._get_py_files(workspace_path):
94+
module_path = self._convert_to_module_path(py_file, workspace_path)
95+
module = importlib.import_module(module_path)
96+
for attr_name in dir(module):
97+
obj = getattr(module, attr_name)
98+
if isinstance(obj, Source):
99+
definitions.sources.add(obj)
100+
elif isinstance(obj, Feature):
101+
definitions.features.add(obj)
102+
elif isinstance(obj, DerivedFeature):
103+
definitions.derived_features.add(obj)
104+
elif isinstance(obj, FeatureAnchor):
105+
definitions.feature_anchors.add(obj)
106+
elif isinstance(obj, Transformation):
107+
definitions.transformations.add(obj)
108+
return definitions
109+
110+
def save_to_feature_config(self, workspace_path: Path, config_save_dir: Path):
111+
"""Save feature definition within the workspace into HOCON feature config files"""
112+
repo_definitions = self._extract_features(workspace_path)
113+
self._save_request_feature_config(repo_definitions, config_save_dir)
114+
self._save_anchored_feature_config(repo_definitions, config_save_dir)
115+
self._save_derived_feature_config(repo_definitions, config_save_dir)
116+
117+
def save_to_feature_config_from_context(self, anchor_list, derived_feature_list, local_workspace_dir: Path):
118+
"""Save feature definition within the workspace into HOCON feature config files from current context, rather than reading from python files"""
119+
repo_definitions = self._extract_features_from_context(
120+
anchor_list, derived_feature_list, local_workspace_dir)
121+
self._save_request_feature_config(repo_definitions, local_workspace_dir)
122+
self._save_anchored_feature_config(repo_definitions, local_workspace_dir)
123+
self._save_derived_feature_config(repo_definitions, local_workspace_dir)
124+
125+
def _save_request_feature_config(self, repo_definitions: RepoDefinitions, local_workspace_dir="./"):
126+
config_file_name = "feature_conf/auto_generated_request_features.conf"
127+
tm = Template(
128+
"""
129+
// THIS FILE IS AUTO GENERATED. PLEASE DO NOT EDIT.
130+
anchors: {
131+
{% for anchor in feature_anchors %}
132+
{% if anchor.source.name == "PASSTHROUGH" %}
133+
{{anchor.to_feature_config()}}
134+
{% endif %}
135+
{% endfor %}
136+
}
137+
"""
138+
)
139+
140+
request_feature_configs = tm.render(
141+
feature_anchors=repo_definitions.feature_anchors)
142+
config_file_path = os.path.join(local_workspace_dir, config_file_name)
143+
write_to_file(content=request_feature_configs,
144+
full_file_name=config_file_path)
145+
146+
@classmethod
147+
def _save_anchored_feature_config(self, repo_definitions: RepoDefinitions, local_workspace_dir="./"):
148+
config_file_name = "feature_conf/auto_generated_anchored_features.conf"
149+
tm = Template(
150+
"""
151+
// THIS FILE IS AUTO GENERATED. PLEASE DO NOT EDIT.
152+
anchors: {
153+
{% for anchor in feature_anchors %}
154+
{% if not anchor.source.name == "PASSTHROUGH" %}
155+
{{anchor.to_feature_config()}}
156+
{% endif %}
157+
{% endfor %}
158+
}
159+
160+
sources: {
161+
{% for source in sources%}
162+
{% if not source.name == "PASSTHROUGH" %}
163+
{{source.to_feature_config()}}
164+
{% endif %}
165+
{% endfor %}
166+
}
167+
"""
168+
)
169+
anchored_feature_configs = tm.render(feature_anchors=repo_definitions.feature_anchors,
170+
sources=repo_definitions.sources)
171+
config_file_path = os.path.join(local_workspace_dir, config_file_name)
172+
write_to_file(content=anchored_feature_configs,
173+
full_file_name=config_file_path)
174+
175+
@classmethod
176+
def _save_derived_feature_config(self, repo_definitions: RepoDefinitions, local_workspace_dir="./"):
177+
config_file_name = "feature_conf/auto_generated_derived_features.conf"
178+
tm = Template(
179+
"""
180+
anchors: {}
181+
derivations: {
182+
{% for derived_feature in derived_features %}
183+
{{derived_feature.to_feature_config()}}
184+
{% endfor %}
185+
}
186+
"""
187+
)
188+
derived_feature_configs = tm.render(
189+
derived_features=repo_definitions.derived_features)
190+
config_file_path = os.path.join(local_workspace_dir, config_file_name)
191+
write_to_file(content=derived_feature_configs,
192+
full_file_name=config_file_path)
193+

0 commit comments

Comments
 (0)