当数据遇见智能：这对"饭搭子"如何炒出商业价值的满汉全席

# 原始销售数据样例（这泥巴得洗洗）
raw_data = [
    {"price": "29.9元", "sales": "约150份", "date": "周三"},
    {"discount": "满100减20", "库存": "紧张"}
]

# 用PySpark做数据SPA
from pyspark.sql import functions as F

df = spark.read.json("s3://raw_sales_data")
clean_df = df.withColumn("price", F.regexp_replace("price", "元", "").cast("float")) \
            .withColumn("sales", F.split("sales", "约|份")[1].cast("int")) \
            .withColumn("discount_rate", F.when(F.col("discount").contains("满100减20"), 0.2))

# 用TensorFlow做个性化推荐
import tensorflow_recommenders as tfrs

class VideoModel(tfrs.Model):
    def __init__(self, user_model, video_model):
        super().__init__()
        self.user_model = user_model # 用户特征提取器
        self.video_model = video_model # 视频特征提取器
        self.task = tfrs.tasks.Retrieval(metrics=tfrs.metrics.FactorizedTopK(
            videos.batch(128).map(self.video_model)
        )

    def compute_loss(self, features, training=False):
        user_embeddings = self.user_model(features["user_id"]) 
        video_embeddings = self.video_model(features["video_id"])
        return self.task(user_embeddings, video_embeddings)

# 加载10亿级用户行为数据
train_data = tf.data.Dataset.load("hdfs://user_behavior_logs").shuffle(10_000_000)

// 用Flink实现实时风控
DataStream<Transaction> transactions = env
    .addSource(new KafkaSource<>(...))
    .keyBy(Transaction::getUserId);

transactions.process(new KeyedProcessFunction<Long, Transaction, Alert>() {
    private transient ValueState<Double> spendingState;

    public void processElement(Transaction tx, Context ctx, Collector<Alert> out) {
        Double lastHourSpending = spendingState.value() == null ? 0.0 : spendingState.value();
        
        // 检测异常消费：1小时累计超过平均值的5倍
        if (tx.amount > userProfile.getHourlyAvg() * 5) {
            out.collect(new Alert("异常消费", tx));
        }
        
        // 更新滑动窗口（最近1小时）
        spendingState.update(lastHourSpending + tx.amount);
        ctx.timerService().registerEventTimeTimer(ctx.timestamp() + 3600_000);
    }

    public void onTimer(long timestamp, OnTimerContext ctx, Collector<Alert> out) {
        spendingState.clear(); // 每小时清空状态
    }
});

# 用用户反馈数据自动优化模型
import pandas as pd
from sklearn.model_selection import train_test_split

# 每天自动收集新数据
new_data = pd.read_parquet(f"/feedback/{datetime.today().strftime('%Y%m%d')}.parquet")
X_train, X_test, y_train, y_test = train_test_split(new_data["text"], new_data["label"])

# 增量训练
model = tf.keras.models.load_model("current_model.h5")
model.fit(X_train, y_train, validation_data=(X_test, y_test))

# 自动AB测试
if model.evaluate(X_test)["accuracy"] > current_acc * 1.01:
    deploy_new_model(model)