Merge pull request #57 from aliyun/feature/fix_make_seq

fix: consider deduplication method when making sequence features

Author: bruceding

dao/feature_view_dao.go

@@ -92,22 +92,53 @@ func makeSequenceFeatures(offlineSequences, onlineSequences []*sequenceInfo, seq
 
 	//produce seqeunce feature correspond to easyrec processor
 	sequencesValueMap := make(map[string][]string)
-	sequenceMap := make(map[string]bool)
+	sequenceMap := make(map[string]struct{})
 
 	for _, seq := range onlineSequences {
-		key := fmt.Sprintf("%s#%s", seq.itemId, seq.event)
+		var key string
+		switch sequenceConfig.DeduplicationMethodNum {
+		case 1:
+			key = fmt.Sprintf("%s#%s", seq.itemId, seq.event)
+		case 2:
+			key = fmt.Sprintf("%s#%s#%d", seq.itemId, seq.event, seq.timestamp)
+		}
 		if _, exist := sequenceMap[key]; !exist {
-			sequenceMap[key] = true
+			sequenceMap[key] = struct{}{}
 			sequencesValueMap[sequenceConfig.ItemIdField] = append(sequencesValueMap[sequenceConfig.ItemIdField], seq.itemId)
 			sequencesValueMap[sequenceConfig.TimestampField] = append(sequencesValueMap[sequenceConfig.TimestampField], fmt.Sprintf("%d", seq.timestamp))
 			sequencesValueMap[sequenceConfig.EventField] = append(sequencesValueMap[sequenceConfig.EventField], seq.event)
 			if sequenceConfig.PlayTimeField != "" {
 				sequencesValueMap[sequenceConfig.PlayTimeField] = append(sequencesValueMap[sequenceConfig.PlayTimeField], fmt.Sprintf("%.2f", seq.playTime))
 			}
 			sequencesValueMap["ts"] = append(sequencesValueMap["ts"], fmt.Sprintf("%d", currTime-seq.timestamp))
-			for k, v := range seq.onlineBehaviourTableFieldsMap {
-				sequencesValueMap[k] = append(sequencesValueMap[k], v)
-			}
+		}
+	}
+
+	properties := make(map[string]interface{})
+	for key, value := range sequencesValueMap {
+		curSequenceSubName := (seqConfig.OnlineSeqName + "__" + key)
+		properties[curSequenceSubName] = strings.Join(value, ";")
+	}
+	properties[seqConfig.OnlineSeqName] = strings.Join(sequencesValueMap[sequenceConfig.ItemIdField], ";")
+
+	return properties
+
+}
+
+func makeSequenceFeatures4FeatureDB(sequencesInfos []*sequenceInfo, seqConfig *api.SeqConfig, sequenceConfig api.FeatureViewSeqConfig, currTime int64) map[string]interface{} {
+	//produce seqeunce feature correspond to easyrec processor
+	sequencesValueMap := make(map[string][]string)
+
+	for _, seq := range sequencesInfos {
+		sequencesValueMap[sequenceConfig.ItemIdField] = append(sequencesValueMap[sequenceConfig.ItemIdField], seq.itemId)
+		sequencesValueMap[sequenceConfig.TimestampField] = append(sequencesValueMap[sequenceConfig.TimestampField], fmt.Sprintf("%d", seq.timestamp))
+		sequencesValueMap[sequenceConfig.EventField] = append(sequencesValueMap[sequenceConfig.EventField], seq.event)
+		if sequenceConfig.PlayTimeField != "" {
+			sequencesValueMap[sequenceConfig.PlayTimeField] = append(sequencesValueMap[sequenceConfig.PlayTimeField], fmt.Sprintf("%.2f", seq.playTime))
+		}
+		sequencesValueMap["ts"] = append(sequencesValueMap["ts"], fmt.Sprintf("%d", currTime-seq.timestamp))
+		for k, v := range seq.onlineBehaviourTableFieldsMap {
+			sequencesValueMap[k] = append(sequencesValueMap[k], v)
 		}
 	}
 

dao/feature_view_featuredb_dao.go

@@ -979,18 +979,14 @@ func (d *FeatureViewFeatureDBDao) GetUserSequenceFeature(keys []interface{}, use
 				eventWg.Add(1)
 				go func(seqConfigs []*api.SeqConfig) {
 					defer eventWg.Done()
-					var onlineSequences []*sequenceInfo
-					var offlineSequences []*sequenceInfo
 
 					// FeatureDB has processed the integration of online sequence features and offline sequence features
 					// Here we put the results into onlineSequences
 
-					if onlineresult := fetchDataFunc(seqConfigs[0].SeqEvent, seqConfigs[0].SeqLen, key, seqConfigsBehaviorFields[0]); onlineresult != nil {
-						onlineSequences = onlineresult
-					}
+					onlineSequences := fetchDataFunc(seqConfigs[0].SeqEvent, seqConfigs[0].SeqLen, key, seqConfigsBehaviorFields[0])
 
 					for _, seqConfig := range seqConfigs {
-						subproperties := makeSequenceFeatures(offlineSequences, onlineSequences, seqConfig, sequenceConfig, currTime)
+						subproperties := makeSequenceFeatures4FeatureDB(onlineSequences, seqConfig, sequenceConfig, currTime)
 						mu.Lock()
 						for k, value := range subproperties {
 							properties[k] = value