美购精排模型

spark/featureEng.py

@@ -282,7 +282,7 @@ def collectFeaturesToDict(samples,columns,prefix):
     idCol = prefix+"id"
     #根据timestamp获取每个user最新的记录
     prefixSamples = samples.groupBy(idCol).agg(F.max("timestamp").alias("timestamp"))
-    resDatas = prefixSamples.join(samples, on=[idCol,"timestamp"], how='left').select(*columns).distinct().collect()
+    resDatas = prefixSamples.join(samples, on=[idCol,"timestamp"], how='inner').select(*columns).distinct().collect()
     return {d[idCol]:json.dumps(d.asDict(),ensure_ascii=False) for d in resDatas}