service model 优化

spark/featureEng2.py

@@ -335,12 +335,16 @@ def itemFeaturesToRedis(samples,itemDF,columns,redisKey):
             conn.set(newKey, v)
             conn.expire(newKey, 60 * 60 * 24 * 7)
 
+    item_static_columns = [col for col in samples.columns if col.endswith("Bucket") or col.endswith("_number")]
+
     #根据timestamp获取每个user最新的记录
     prefixSamples = samples.groupBy(idCol).agg(F.max("timestamp").alias(timestampCol))
-    item_static_columns = [col for col in prefixSamples.columns if col.endswith("Bucket") or col.endswith("_number")]
-    prefixSamples = prefixSamples.select(*item_static_columns)
-    prefixSamples.show(10,truncate=False)
-    resDatas = itemDF.join(prefixSamples, on=[idCol], how='left').where(F.col("timestamp") == F.col(timestampCol))
+    item_static_df = samples.join(prefixSamples, on=[idCol], how='left').where(F.col("timestamp") == F.col(timestampCol))
+    item_static_df = item_static_df.select(*item_static_columns)
+    item_static_df.show(10,truncate=False)
+
+    resDatas = itemDF.join(item_static_df, on=[idCol], how='left')
+
     for col in item_static_columns:
         res = "0"
         if col.endswith("Bucket"):
@@ -835,7 +839,7 @@ if __name__ == '__main__':
     # model columns
     print("model columns to redis...")
     model_columns = user_columns + item_columns
-    # featureColumnsToRedis(model_columns)
+    featureColumnsToRedis(model_columns)
 
     print("数据字典save...")
     print("dataVocab:", str(dataVocab.keys()))