-
Notifications
You must be signed in to change notification settings - Fork 43
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Fix cache key collision when 2 tables have same list of entity (#187)
* Change in-memory cache options constructor arguments * Restructure feature retrievaer initialisation * Restructure feature retriever initialisation * Restructure feature retriever initialisation * Rework feature cache * Rework feature cache * Handle different column ordering in feature cache * Update test after reworking cache * Move helper functions into utils.go and converter.go * Move internal type into types.go * Move Benchmark_buildEntitiesRequest_geohashArrays to feature_retriever_bench_test.go * Move batch call into batch_call.go * Move extract feast value test to converter_test.go * Add unit test got GetTableName * Add batch_call_test.go * Use entitySet instead of entityIndexMap * Add ToInt32 conversion * Ensure ordering should be maintained in feature cache * Remove debugging code * Refactor feature_retriever.go * Refactor batch_call.go * Rename batch call to call * Move mergeColumnTypes to types.go
- Loading branch information
aria
authored
Oct 5, 2021
1 parent
b892e5b
commit 9a98e57
Showing
20 changed files
with
2,187 additions
and
2,323 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,144 @@ | ||
| package feast | ||
|
|
||
| import ( | ||
| "context" | ||
| "fmt" | ||
| "time" | ||
|
|
||
| feast "github.com/feast-dev/feast/sdk/go" | ||
| "github.com/feast-dev/feast/sdk/go/protos/feast/serving" | ||
| "github.com/feast-dev/feast/sdk/go/protos/feast/types" | ||
| "github.com/opentracing/opentracing-go" | ||
| "go.uber.org/zap" | ||
|
|
||
| "github.com/gojek/merlin/pkg/transformer/spec" | ||
| transTypes "github.com/gojek/merlin/pkg/transformer/types" | ||
| "github.com/gojek/merlin/pkg/transformer/types/converter" | ||
| ) | ||
|
|
||
| type call struct { | ||
| featureTableSpec *spec.FeatureTable | ||
| columns []string | ||
| entitySet map[string]bool | ||
| defaultValues defaultValues | ||
|
|
||
| feastClient feast.Client | ||
| feastURL string | ||
|
|
||
| logger *zap.Logger | ||
|
|
||
| statusMonitoringEnabled bool | ||
| valueMonitoringEnabled bool | ||
| } | ||
|
|
||
| // do create request to feast and return the result as table | ||
| func (fc *call) do(ctx context.Context, entityList []feast.Row, features []string) callResult { | ||
| tableName := GetTableName(fc.featureTableSpec) | ||
|
|
||
| span, ctx := opentracing.StartSpanFromContext(ctx, "feast.doBatchCall") | ||
| span.SetTag("feast.url", fc.feastURL) | ||
| span.SetTag("table", tableName) | ||
| defer span.Finish() | ||
|
|
||
| feastRequest := feast.OnlineFeaturesRequest{ | ||
| Project: fc.featureTableSpec.Project, | ||
| Entities: entityList, | ||
| Features: features, | ||
| } | ||
|
|
||
| startTime := time.Now() | ||
| feastResponse, err := fc.feastClient.GetOnlineFeatures(ctx, &feastRequest) | ||
| durationMs := time.Now().Sub(startTime).Milliseconds() | ||
| if err != nil { | ||
| feastLatency.WithLabelValues("error", fc.feastURL).Observe(float64(durationMs)) | ||
| feastError.WithLabelValues(fc.feastURL).Inc() | ||
|
|
||
| return callResult{featureTable: nil, err: err} | ||
| } | ||
| feastLatency.WithLabelValues("success", fc.feastURL).Observe(float64(durationMs)) | ||
|
|
||
| featureTable, err := fc.processResponse(feastResponse) | ||
| if err != nil { | ||
| return callResult{featureTable: nil, err: err} | ||
| } | ||
|
|
||
| return callResult{tableName: tableName, featureTable: featureTable, err: nil} | ||
| } | ||
|
|
||
| // processResponse process response from feast serving and create an internal feature table representation of it | ||
| func (fc *call) processResponse(feastResponse *feast.OnlineFeaturesResponse) (*internalFeatureTable, error) { | ||
| responseStatus := feastResponse.Statuses() | ||
| responseRows := feastResponse.Rows() | ||
| entities := make([]feast.Row, len(responseRows)) | ||
| valueRows := make([]transTypes.ValueRow, len(responseRows)) | ||
| columnTypes := make([]types.ValueType_Enum, len(fc.columns)) | ||
|
|
||
| for rowIdx, feastRow := range responseRows { | ||
| valueRow := make(transTypes.ValueRow, len(fc.columns)) | ||
|
|
||
| // create entity object, for cache key purpose | ||
| entity := feast.Row{} | ||
| for colIdx, column := range fc.columns { | ||
| var rawValue *types.Value | ||
|
|
||
| featureStatus := responseStatus[rowIdx][column] | ||
| switch featureStatus { | ||
| case serving.GetOnlineFeaturesResponse_PRESENT: | ||
| rawValue = feastRow[column] | ||
| // set value of entity | ||
| _, isEntity := fc.entitySet[column] | ||
| if isEntity { | ||
| entity[column] = rawValue | ||
| } | ||
| case serving.GetOnlineFeaturesResponse_NOT_FOUND, serving.GetOnlineFeaturesResponse_NULL_VALUE, serving.GetOnlineFeaturesResponse_OUTSIDE_MAX_AGE: | ||
| defVal, ok := fc.defaultValues.GetDefaultValue(fc.featureTableSpec.Project, column) | ||
| if !ok { | ||
| // no default value is specified, we populate with nil | ||
| valueRow[colIdx] = nil | ||
| continue | ||
| } | ||
| rawValue = defVal | ||
| default: | ||
| return nil, fmt.Errorf("unsupported feature retrieval status: %s", featureStatus) | ||
| } | ||
|
|
||
| val, valType, err := converter.ExtractFeastValue(rawValue) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| // if previously we detected that the column type is invalid then we set it to the correct type | ||
| if valType != types.ValueType_INVALID { | ||
| columnTypes[colIdx] = valType | ||
| } | ||
| valueRow[colIdx] = val | ||
|
|
||
| fc.recordMetrics(val, column, featureStatus) | ||
| } | ||
|
|
||
| entities[rowIdx] = entity | ||
| valueRows[rowIdx] = valueRow | ||
| } | ||
|
|
||
| return &internalFeatureTable{ | ||
| entities: entities, | ||
| columnNames: fc.columns, | ||
| columnTypes: columnTypes, | ||
| valueRows: valueRows, | ||
| }, nil | ||
| } | ||
|
|
||
| func (fc *call) recordMetrics(val interface{}, column string, featureStatus serving.GetOnlineFeaturesResponse_FieldStatus) { | ||
| // put behind feature toggle since it will generate high cardinality metrics | ||
| if fc.valueMonitoringEnabled { | ||
| v, err := converter.ToFloat64(val) | ||
| if err == nil { | ||
| feastFeatureSummary.WithLabelValues(column).Observe(v) | ||
| } | ||
| } | ||
|
|
||
| // put behind feature toggle since it will generate high cardinality metrics | ||
| if fc.statusMonitoringEnabled { | ||
| feastFeatureStatus.WithLabelValues(column, featureStatus.String()).Inc() | ||
| } | ||
| } |
Oops, something went wrong.