3. 数据采集与接入：用户行为数据采集（埋点、日志）、业务数据接入（CRM、订单）、数据清洗与预处理

好，咱们进入实战环节的第一个硬骨头——数据采集。

说实话，很多做标签系统的团队，最后翻车都翻在数据采集上。不是技术多难，而是从一开始就没想清楚：我要采什么？怎么采？采完了怎么用？

我个人习惯，在动手写一行代码之前，先画一张数据流向图。你想想看，用户行为数据、业务数据，这两条河最终要汇入同一个湖里。如果源头的水质不行，后面再怎么洗也白搭。

3.1 用户行为数据采集：埋点与日志

用户行为数据，说白了就是记录「谁在什么时间、什么地方、做了什么事」。比如点击、浏览、搜索、加购、下单。

这里有两个主流方案：前端埋点和后端日志。我建议你两个都要，因为各有各的用处。

3.1.1 前端埋点

前端埋点又分三种：代码埋点、可视化埋点、无埋点。

  • 代码埋点：最灵活，想采什么就采什么。但开发成本高，每次加个按钮都要发版。
  • 可视化埋点：运营同学在页面上圈选元素，自动生成埋点。方便，但复杂交互搞不定。
  • 无埋点：全量采集所有点击事件。数据量大，但噪音也大。

我在项目中遇到过最坑的事：运营同学用可视化埋点圈了一个「立即购买」按钮，结果页面改版后按钮的CSS类名变了，埋点失效了整整两周。嗯，从那以后我定了个规矩——核心转化路径必须用代码埋点。

核心原则：代码埋点保核心，可视化埋点做探索，无埋点做兜底。

来看一个简单的代码埋点示例：

// 前端埋点示例 - 点击事件上报
function trackEvent(eventName, properties = {}) {
    const payload = {
        event: eventName,
        user_id: getUserId(),          // 从cookie或localStorage获取
        timestamp: Date.now(),
        page: window.location.pathname,
        properties: properties,
        device: navigator.userAgent
    };
    
    // 使用sendBeacon，页面关闭时也能上报
    navigator.sendBeacon('/api/track', JSON.stringify(payload));
}

// 使用方式
document.getElementById('buy-btn').addEventListener('click', function() {
    trackEvent('click_buy_now', {
        product_id: 'P10086',
        price: 299.00,
        from: 'detail_page'
    });
});

小技巧：用 navigator.sendBeacon 代替 fetch 或 XMLHttpRequest。它不阻塞页面卸载，数据不会丢。我曾经因为这个改动，把下单页面的埋点丢失率从 15% 降到了 0.5%。

3.1.2 后端日志采集

前端埋点能记录「用户点了什么」，但后端日志能告诉你「系统处理了什么」。比如用户下单，前端只能看到点击事件，后端日志能看到订单是否创建成功、库存是否扣减、支付是否完成。

我建议用 JSON 格式的日志，别再用那种用竖线分隔的古老格式了。JSON 可读性强，解析方便，直接怼到 Elasticsearch 里就能查。

# 后端日志示例 - Python logging + JSON格式
import logging
import json
from datetime import datetime

class JsonFormatter(logging.Formatter):
    def format(self, record):
        log_record = {
            "timestamp": datetime.utcnow().isoformat(),
            "level": record.levelname,
            "module": record.module,
            "function": record.funcName,
            "message": record.getMessage(),
            "trace_id": getattr(record, 'trace_id', None),  # 链路追踪ID
            "user_id": getattr(record, 'user_id', None),
            "request_id": getattr(record, 'request_id', None)
        }
        return json.dumps(log_record)

# 配置日志
logger = logging.getLogger('user_behavior')
handler = logging.FileHandler('/var/log/app/behavior.log')
handler.setFormatter(JsonFormatter())
logger.addHandler(handler)

# 使用
logger.info('order_created', extra={
    'user_id': 'U12345',
    'trace_id': 'abc-def-ghi',
    'order_id': 'O20240315001',
    'amount': 299.00
})

注意：千万别在日志里打印用户密码、身份证号、手机号等敏感信息。我曾经见过一个团队，把用户明文密码打到了日志里，结果日志文件被运维同学不小心传到了公开的S3桶上……嗯，那场面，法务部直接介入。

3.2 业务数据接入：CRM与订单

用户行为数据是「态度」，业务数据是「结果」。你光知道用户浏览了商品页（行为），但不知道他到底买没买（业务），那标签就是瘸腿的。

业务数据通常来自：

  • CRM系统：用户基本信息、会员等级、积分、客服记录
  • 订单系统：购买记录、退款记录、支付方式、收货地址
  • 商品系统：浏览过的商品类目、品牌、价格区间

接入方式我推荐两种：

方式 适用场景 延迟 我踩过的坑
实时API 用户注册、下单等核心事件 秒级 API限流没处理好，高峰期把CRM打挂了
离线批量同步 历史数据、非实时报表 T+1 没做增量同步，每天全量拉取，数据量大了之后跑不动

我个人习惯：核心业务用实时API + 消息队列，非核心用离线批量。举个例子，用户下单这个事件，必须实时同步到标签系统，因为我们要立刻更新「近30天购买次数」这个标签。而用户的历史订单，每天凌晨跑一次批量同步就够了。

# 实时接入示例 - 通过Kafka消费订单事件
from kafka import KafkaConsumer
import json

consumer = KafkaConsumer(
    'order_events',
    bootstrap_servers=['kafka1:9092', 'kafka2:9092'],
    group_id='tag_system_group',
    value_deserializer=lambda m: json.loads(m.decode('utf-8'))
)

for message in consumer:
    order = message.value
    user_id = order['user_id']
    
    # 更新用户标签
    update_tag(user_id, 'last_order_time', order['created_at'])
    update_tag(user_id, 'total_order_amount', order['amount'], agg='sum')
    update_tag(user_id, 'order_count', 1, agg='count')
    
    print(f"已处理用户 {user_id} 的订单 {order['order_id']}")

3.3 数据清洗与预处理

数据采进来了，但别高兴太早。真实世界的数据，脏到你怀疑人生。

我总结了一套「清洗三板斧」：

  1. 去重：同一个用户因为网络重试，上报了两次「注册」事件
  2. 补全：用户ID为空、时间戳格式不对、设备型号缺失
  3. 标准化：有的系统用「男/女」，有的用「1/0」，有的用「M/F」

一个血泪教训：我曾经接手过一个项目，CRM里用户性别字段，有的存「男」，有的存「先生」，有的存「male」，还有的存「1」。三个系统，四种写法。最后清洗脚本写了200行，全是if-else。

来看一个清洗示例：

# 数据清洗示例 - PySpark处理
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, when, udf
from pyspark.sql.types import StringType

spark = SparkSession.builder.appName("data_cleaning").getOrCreate()

# 读取原始数据
df = spark.read.json("hdfs://data/raw/user_behavior/2024-03-15/")

# 1. 去重：按事件ID去重，保留第一条
df_clean = df.dropDuplicates(["event_id"])

# 2. 补全：用户ID为空的行直接过滤
df_clean = df_clean.filter(col("user_id").isNotNull())

# 3. 标准化：统一性别字段
def standardize_gender(gender):
    mapping = {
        "男": "male", "先生": "male", "M": "male", "1": "male",
        "女": "female", "女士": "female", "F": "female", "0": "female"
    }
    return mapping.get(str(gender).strip(), "unknown")

standardize_gender_udf = udf(standardize_gender, StringType())
df_clean = df_clean.withColumn("gender", standardize_gender_udf(col("gender")))

# 4. 时间格式统一
df_clean = df_clean.withColumn(
    "event_time",
    when(col("event_time").rlike("^\d{4}-\d{2}-\d{2}"), col("event_time"))
    .otherwise(to_timestamp(col("event_time"), "yyyy/MM/dd HH:mm:ss"))
)

df_clean.write.parquet("hdfs://data/clean/user_behavior/2024-03-15/")

我的习惯：清洗后的数据，一定要写一个数据质量报告。比如：原始数据100万条，去重后剩98万条，过滤空ID后剩95万条。这样你心里有数，也知道哪里需要优化。

最后说一句：数据清洗不是一次性工作。业务变了、系统升级了、新渠道接入了，脏数据的「花样」也会变。我建议每周跑一次数据质量监控，自动报警。别等到做标签的时候才发现数据有问题——那时候已经晚了。

← 上一章 📖 返回目录 下一章 →