Visitar URL original
feat: Local compute engine by haoxu0 · Pull Request #5278 · feast-dev/feast · GitHub
Skip to content

feat: Local compute engine #5278

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
franciscojavierarceo merged 9 commits into from
Apr 18, 2025
Merged

feat: Local compute engine #5278

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
ce6b022
Create Local Compute Engine
haoxu0 Apr 16, 2025
daa8f55
format code
haoxu0 Apr 16, 2025
d5f847a
format code
haoxu0 Apr 16, 2025
107cc33
format code
haoxu0 Apr 16, 2025
2dd267b
update backend
haoxu0 Apr 17, 2025
4b801ba
update backend
haoxu0 Apr 17, 2025
f254b80
update status
haoxu0 Apr 17, 2025
f89ebb1
update doc
haoxu0 Apr 17, 2025
d6f0ed3
format
haoxu0 Apr 17, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
format 
Signed-off-by: HaoXuAI <sduxuhao@gmail.com>
  • Loading branch information
@haoxu0
haoxu0 committed Apr 17, 2025
commit d6f0ed3506a4fb9a76f9cb710c3541fc7c82dce6
27 changes: 10 additions & 17 deletions sdk/python/feast/infra/compute_engines/local/feature_builder.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,9 +19,9 @@

class LocalFeatureBuilder(FeatureBuilder):
def __init__(
self,
task: Union[MaterializationTask, HistoricalRetrievalTask],
backend: DataFrameBackend,
self,
task: Union[MaterializationTask, HistoricalRetrievalTask],
backend: DataFrameBackend,
):
super().__init__(task)
self.backend = backend
Expand All @@ -31,15 +31,13 @@ def build_source_node(self):
self.nodes.append(node)
return node

def build_join_node(self,
input_node):
def build_join_node(self, input_node):
node = LocalJoinNode("join", self.backend)
node.add_input(input_node)
self.nodes.append(node)
return node

def build_filter_node(self,
input_node):
def build_filter_node(self, input_node):
filter_expr = None
if hasattr(self.feature_view, "filter"):
filter_expr = self.feature_view.filter
Expand All @@ -62,8 +60,7 @@ def _get_aggregate_operations(agg_specs):
agg_ops[alias] = (agg.function, agg.column)
return agg_ops

def build_aggregation_node(self,
input_node):
def build_aggregation_node(self, input_node):
agg_specs = self.feature_view.aggregations
agg_ops = self._get_aggregate_operations(agg_specs)
group_by_keys = self.feature_view.entities
Expand All @@ -72,33 +69,29 @@ def build_aggregation_node(self,
self.nodes.append(node)
return node

def build_dedup_node(self,
input_node):
def build_dedup_node(self, input_node):
node = LocalDedupNode("dedup", self.backend)
node.add_input(input_node)
self.nodes.append(node)
return node

def build_transformation_node(self,
input_node):
def build_transformation_node(self, input_node):
node = LocalTransformationNode(
"transform", self.feature_view.feature_transformation, self.backend
)
node.add_input(input_node)
self.nodes.append(node)
return node

def build_validation_node(self,
input_node):
def build_validation_node(self, input_node):
node = LocalValidationNode(
"validate", self.feature_view.validation_config, self.backend
)
node.add_input(input_node)
self.nodes.append(node)
return node

def build_output_nodes(self,
input_node):
def build_output_nodes(self, input_node):
node = LocalOutputNode("output")
node.add_input(input_node)
self.nodes.append(node)
Expand Down
4 changes: 1 addition & 3 deletions sdk/python/feast/infra/compute_engines/spark/feature_builder.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -56,9 +56,7 @@ def build_filter_node(self, input_node):
if hasattr(self.feature_view, "filter"):
filter_expr = self.feature_view.filter
ttl = self.feature_view.ttl
node = SparkFilterNode(
"filter", self.spark_session, ttl, filter_expr
)
node = SparkFilterNode("filter", self.spark_session, ttl, filter_expr)
node.add_input(input_node)
self.nodes.append(node)
return node
Expand Down