/
feature_view_redis_dao.go
93 lines (80 loc) · 2.38 KB
/
feature_view_redis_dao.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
package dao
import (
"context"
"fmt"
"strings"
"github.com/go-redis/redis/v8"
"github.com/aliyun/aliyun-pai-featurestore-go-sdk/constants"
"github.com/aliyun/aliyun-pai-featurestore-go-sdk/datasource/redisdb"
"github.com/aliyun/aliyun-pai-featurestore-go-sdk/utils"
)
type FeatureViewRedisDao struct {
redisClinet *redis.Client
redisPrefix string
redisDelimiter string
primaryKeyField string
eventTimeField string
ttl int
fields []string
fieldTypeMap map[string]constants.FSType
fieldIndexMap map[string]int
}
func NewFeatureViewRedisDao(config DaoConfig) *FeatureViewRedisDao {
dao := FeatureViewRedisDao{
redisPrefix: config.RedisPrefix,
redisDelimiter: "\u0001",
primaryKeyField: config.PrimaryKeyField,
eventTimeField: config.EventTimeField,
ttl: config.TTL,
fields: config.Fields,
fieldTypeMap: config.FieldTypeMap,
fieldIndexMap: make(map[string]int, len(config.Fields)),
}
client, err := redisdb.GetRedis(config.RedisName)
if err != nil {
return nil
}
dao.redisClinet = client.GetClient()
for i, field := range dao.fields {
dao.fieldIndexMap[field] = i
}
return &dao
}
func (d *FeatureViewRedisDao) GetFeatures(keys []interface{}, selectFields []string) ([]map[string]interface{}, error) {
var pkeys []string
for _, key := range keys {
pkey := utils.ToString(key, "")
pkeys = append(pkeys, fmt.Sprintf("%s%s", d.redisPrefix, pkey))
}
ctx := context.Background()
values, err := d.redisClinet.MGet(ctx, pkeys...).Result()
if err != nil {
return nil, err
}
result := make([]map[string]interface{}, 0, len(keys))
for i, value := range values {
if value == nil {
result = append(result, map[string]interface{}{d.primaryKeyField: keys[i]})
} else {
if str, ok := value.(string); ok {
m := make(map[string]interface{}, len(selectFields)+1)
m[d.primaryKeyField] = keys[i]
strs := strings.Split(str, d.redisDelimiter)
for _, field := range selectFields {
if index, ok := d.fieldIndexMap[field]; ok {
switch d.fieldTypeMap[field] {
case constants.FS_DOUBLE, constants.FS_FLOAT:
m[field] = utils.ToFloat(strs[index], 0)
case constants.FS_INT32, constants.FS_INT64:
m[field] = utils.ToInt(strs[index], 0)
default:
m[field] = strs[index]
}
}
}
result = append(result, m)
}
}
}
return result, nil
}