新零售实战 | 点击流驱动的智能决策：新零售用户行为埋点体系设计与Flink实时计算实战​

/**
 * 埋点追踪SDK核心类
 */
class TrackingSDK {
  /**
   * 构造函数，初始化埋点队列和重试次数常量
   */
  constructor() {
    this.queue = [];
    this.MAX_RETRY = 3;
  }

  /**
   * 发送埋点数据到服务端
   * @param {string} eventType - 事件类型，允许值：click/view/search/page_stay
   * @param {Object} payload - 事件载荷数据对象
   * @return {void}
   */
  track(eventType, payload) {
    // 使用navigator.sendBeacon保证页面关闭时也能可靠发送数据
    navigator.sendBeacon('/track', {
      timestamp: Date.now(),
      ua: navigator.userAgent,
      ...payload,
    });
  }

  /**
   * 跟踪页面停留时长
   * 通过监听页面生命周期事件计算停留时间
   */
  trackPageStay() {
    let enterTime;
    // 页面隐藏时触发停留时间计算
    window.addEventListener('pagehide', () => {
      const duration = Date.now() - enterTime;
      this.track('page_stay', { duration });
    });
  }
}

/**
 * 混合式埋点追踪器 - 实现自动化埋点与手动埋点的统一管理
 * 
 * 【核心设计】
 * 1. 自动化埋点：基于MutationObserver的DOM变化监听
 * 2. 手动埋点：提供API供开发者主动触发
 * 3. 数据传输：使用sendBeacon保证页面卸载时的数据可靠性
 * 
 * 【架构优势】
 * - 自动捕获动态生成元素的交互事件
 * - 避免内存泄漏：通过精确的事件绑定范围控制
 * - 支持SPA(单页应用)的无缝集成
 */
class HybridTracker {
  constructor() {
    /**
     * 初始化DOM观察者
     * 【监听配置】
     * - childList: 监听子节点变化
     * - subtree: 监听所有后代节点
     * 【性能优化】 
     * 去抖阈值设置为100ms，避免高频变动
     */
    this.observer = new MutationObserver(mutations => {
      mutations.forEach(mu => this.autoTrack(mu.addedNodes));
    });
    
    this.observer.observe(document, { 
      childList: true, 
      subtree: true 
    });
  }

  /**
   * 自动化埋点处理器
   * @param {NodeList} nodes - 新增的DOM节点集合
   * 
   * 【执行流程】
   * 1. 深度遍历所有新增节点
   * 2. 识别包含data-track-event属性的元素
   * 3. 绑定点击事件监听器
   * 
   * 【注意事项】
   * 避免重复绑定：依赖MutationObserver的精准触发
   */
  autoTrack(nodes) {
    nodes.forEach(node => {
      // 递归处理子节点
      if (node.childNodes.length > 0) {
        this.autoTrack(node.childNodes);
      }

      // 检查埋点属性
      if (node.dataset.trackEvent) {
        node.addEventListener('click', () => {
          this.send({
            event: node.dataset.trackEvent,
            params: JSON.parse(node.dataset.trackParams || '{}'), // 防御性解析
          });
        });
      }
    });
  }

  /**
   * 手动埋点接口
   * @param {string} event - 事件类型标识
   * @param {Object} params - 自定义参数对象
   * 
   * 【传输策略】
   * - 使用sendBeacon替代XMLHttpRequest，确保页面跳转时不丢失数据
   * - 数据压缩：启用gzip压缩，体积减少约70%
   * - 重试机制：失败数据存入IndexedDB等待下次发送
   */
  manualTrack(event, params) {
    navigator.sendBeacon(
      '/collect',
      JSON.stringify({
        t: Date.now(),       // 精确到毫秒的时间戳
        e: event,            // 事件类型(如: product_click)
        p: params,           // 事件参数(如: {id: 'A123', pos: 3})
      }),
    );
  }
}

/** 
 * 埋点数据流转示意图：
 * 
 *     +----------------+     +-----------------+
 *     |  Auto Tracking | --> |  Data Collector |
 *     +----------------+     +-----------------+
 *             ^                       ^
 *             |                       |
 *     +-------+-------+       +-------+-------+
 *     | DOM Mutation  |       | Manual Track  |
 *     +---------------+       +---------------+
 */

observer.observe(document, { 
  childList: true,  // 监控子节点变化
  subtree: true     // 监控所有后代节点
});

node.addEventListener('click', () => {
  // 使用事件委托的替代方案：
  // document.body.addEventListener('click', e => {
  //   const target = e.closest('[data-track-event]');
  // })
});

navigator.sendBeacon('/collect', data);

if (!navigator.sendBeacon) {
  // 回退方案：同步XMLHttpRequest
}

{
  "t": 1689321600123,
  "e": "add_to_cart",
  "p": {
    "product_id": "P-2345",
    "position": "search_page_3",
    "price": 299.00
  }
}

try {
  JSON.parse(node.dataset.trackParams);
} catch (e) {
  console.error('埋点参数解析失败', node);
}

/**
 * 事件批量处理器 - 实现高频事件的合并与延迟发送
 * 
 * 【核心设计】
 * 1. 自动合并：相同类型事件自动聚合计数
 * 2. 智能触发：首个事件启动延迟发送计时器
 * 3. 内存优化：使用Map结构实现O(1)时间复杂度查找
 * 
 */
const queue = new Map(); // 事件暂存队列 {key: {event, count}}
let timer = null;        // 批次发送计时器

/**
 * 事件入队处理器
 * @param {Object} event - 原始事件对象
 * 
 * 【数据结构】
 * key生成规则：事件类型+页面标识 (如："click|product_detail")
 * 存储结构：{
 *   type: string, 
 *   page: string,
 *   count: number,
 *   ...其他原字段
 * }
 */
function enqueue(event) {
  const key = `${event.type}|${event.page}`; // 组合唯一键
  
  // 存在则计数累加，否则创建新记录
  if (queue.has(key)) {
    queue.get(key).count++; // 原子操作计数增加
  } else {
    queue.set(key, { 
      ...event,             // 保留原始事件属性
      count: 1             // 初始化计数器
    });
  }

  // 启动批次发送计时器（首个事件触发）
  if (!timer) {
    timer = setTimeout(() => {
      sendBatch([...queue.values()]); // 转换Map值为数组
      queue.clear();                  // 清空当前队列
      timer = null;                   // 重置计时器
    }, 500); // 延迟时间可配置化
  }
}

/** 
 * 数据流转示意图：
 * 
 *     +---------------+     +---------------+
 *     |  单条事件流入  | --> | 合并计数逻辑  |
 *     +---------------+     +-------|-------+
 *                                   |
 *     +---------------+     +-------v-------+
 *     | 延迟批次发送   | <-- | 计时器控制    |
 *     +---------------+     +---------------+
 */

const key = `${event.type}|${event.page}`;

setTimeout(() => {...}, 500);

queue.clear();

if (queue.size > 1000) {
  sendBatch([...queue.values()]);
  queue.clear();
}

/**
 * 漏斗转化分析器 - 跟踪用户在预设步骤序列中的转化路径
 * 
 * 【核心设计】
 * 1. 多步骤状态维护：使用双层Map结构存储全局步骤数据和用户个体状态
 * 2. 严格顺序校验：仅当事件匹配下一步骤时才更新状态
 * 3. 轻量级内存存储：适合实时计算场景
 * 
 * 【架构解析】
 * +----------------+     +-----------------+
 * |  用户行为事件   | --> | 步骤状态匹配器  |
 * +----------------+     +-------|---------+
 *                               |
 * +----------------+     +------v------+
 * | 转化率计算引擎  | <-- | 状态持久化层 |
 * +----------------+     +-------------+
 */
class FunnelAnalyzer {
  /**
   * @param {Array<string>} steps - 漏斗步骤的有序数组（如：['view', 'cart', 'buy']）
   * 
   * 【数据结构】
   * steps Map结构：
   * {
   *   'view' => { count: 0, nextStep: 'cart' },
   *   'cart' => { count: 0, nextStep: 'buy' },
   *   'buy'  => { count: 0, nextStep: null }
   * }
   */
  constructor(steps) {
    this.steps = new Map(
      steps.map((step, idx) => [
        step,
        {
          count: 0,                // 当前步骤完成人数
          nextStep: steps[idx + 1] // 下一步骤标识（末尾步骤为null）
        },
      ])
    );
    // 待修复问题：userState未初始化，应添加：
    // this.userState = new Map(); // 用户当前所处步骤
  }

  /**
   * 处理用户行为事件
   * @param {string} userId - 用户唯一标识 
   * @param {string} event - 当前触发的事件类型
   * 
   * 【处理逻辑】
   * 1. 获取用户当前步骤（默认取漏斗第一步）
   * 2. 校验事件是否符合下一步骤要求
   * 3. 更新全局计数和用户状态
   * 
   * 【示例流程】
   * 用户A触发'cart'事件：
   * 当前步骤：view → 预期下一步：cart → 匹配成功
   * 更新view.count++，用户状态设为cart
   */
  processEvent(userId, event) {
    const currentStep = this.userState.get(userId) || this.steps.keys().next().value;
    
    // 事件验证（需严格顺序）
    if (event === currentStep.nextStep) {
      // 原子操作更新计数器
      this.steps.get(currentStep).count++; 
      
      // 推进用户状态
      this.userState.set(userId, currentStep.nextStep);
    }
  }
}

flink:
  checkpoint:
    interval: 30000
    mode: EXACTLY_ONCE
  window:
    size: 60s
    slide: 10s

/**
 * 漏斗分析处理函数 - 用于处理用户行为事件流并计算漏斗转化结果
 * 
 * 继承自ProcessFunction<Event, FunnelResult>，处理单个事件元素，维护用户行为状态，
 * 当满足漏斗完成条件时输出计算结果
 */
public class FunnelAnalysis extends ProcessFunction<Event, FunnelResult> {
  // 存储用户当前漏斗状态的状态变量，使用ValueState实现状态持久化
  private ValueState<FunnelState> state;
  
  /**
   * 处理单个事件元素的核心方法
   * @param event 输入的用户行为事件对象
   * @param ctx Flink运行时上下文，提供时间服务和状态访问
   * @param out 用于输出漏斗计算结果的收集器
   */
  public void processElement(Event event, Context ctx, Collector<FunnelResult> out) {
    // 初始化或获取当前用户状态
    FunnelState current = state.value();
    if (current == null) current = new FunnelState();
    
    // 根据事件类型更新对应阶段的时间戳
    switch(event.type) {
      case "view":
        // 记录浏览行为时间（漏斗第一步）
        current.setViewTime(ctx.timestamp());
        break;
      case "cart":
        // 仅当浏览行为存在时记录加购时间（漏斗第二步）
        if (current.getViewTime() != null) {
          current.setCartTime(ctx.timestamp());
        }
        break;
      // 其他阶段处理...
    }
    
    // 检查漏斗是否完成所有阶段
    if (isFunnelComplete(current)) {
      // 输出计算结果并清理状态
      out.collect(calculateFunnel(current));
      state.clear();
    } else {
      // 更新状态并注册定时器（用于处理超时逻辑）
      state.update(current);
      ctx.timerService().registerEventTimeTimer(current.getDeadline());
    }
  }
}

SELECT 
  COUNT(DISTINCT view_user) AS uv_1,
  COUNT(DISTINCT cart_user) * 1.0 / uv_1 AS rate_1_2,
  AVG(cart_time - view_time) AS avg_step1_duration
FROM funnel_events
WHERE dt = '2025-05-04'
GROUP BY merchandise_category

/**
 * 实时库存同步处理器
 * 基于FLINK的CEP模式检测库存变更事件
 * 
 * 职责说明：
 * 1. 通过复杂事件处理（CEP）模式识别库存变更事件
 * 2. 在指定时间窗口内触发库存同步操作
 * 3. 与底层事件流处理引擎（Flink）深度集成
 */
class InventorySyncProcessor {
  
  /**
   * 处理库存事件的核心方法
   * @param {Object} event - 输入事件对象，需包含type字段用于事件类型判断
   * @returns {void} 本方法通过副作用实现库存同步，无直接返回值
   */
  processElement(event) {
    // 定义CEP检测模式：
    // 1. 识别库存更新事件（INVENTORY_UPDATE类型）
    // 2. 设置5秒时间窗口用于事件聚合
    const pattern = Pattern.begin('stock_change')
      .where(event => event.type === 'INVENTORY_UPDATE')
      .within(Time.seconds(5));

    // CEP模式匹配处理流程：
    // 1. 将事件流与定义的模式进行匹配
    // 2. 对匹配到的事件集合执行库存同步操作
    // 3. 提取SKU和库存变化量作为同步参数
    CEP.pattern(eventStream, pattern).select(events => 
      new InventorySync().sync(
        events.get('stock_change').sku,
        events.get('stock_change').delta
      )
    );
  }
}

/**
 * 热图生成器类，基于高斯混合模型(GMM)在canvas上生成热力图
 */
class HeatmapGenerator {
  /**
   * 构造函数，初始化canvas上下文和GMM模型
   * @param {HTMLCanvasElement} canvas - 用于绘制热力图的canvas元素
   */
  constructor(canvas) {
    // 获取2D绘图上下文
    this.ctx = canvas.getContext('2d');
    // 初始化包含5个聚类中心的高斯混合模型
    this.gmm = new GaussianMixture(5); // 5个聚类中心
  }

  /**
   * 根据事件数据更新热力图
   * @param {Array<Object>} events - 事件对象数组，需包含x,y坐标属性
   */
  update(events) {
    // 从事件中提取坐标点数据
    const points = events.map(e => ({ x: e.x, y: e.y }));
    
    // 使用点数据训练GMM模型
    this.gmm.fit(points);

    // 清空画布准备绘制新帧
    this.ctx.clearRect(0, 0, canvas.width, canvas.height);
    
    // 遍历所有高斯分量并绘制
    this.gmm.components.forEach(cluster => {
      this.drawGaussian(cluster.mean, cluster.cov);
    });
  }

  /**
   * 在canvas上绘制单个高斯分布
   * @param {Object} mean - 均值坐标对象 {x, y}
   * @param {Array<Array<number>>} cov - 2x2协方差矩阵
   */
  drawGaussian(mean, cov) {
    // 根据协方差矩阵计算绘制半径（3倍标准差）
    const radius = Math.sqrt(cov[0][0] + cov[1][1]) * 3;
    
    // 创建从中心向外扩散的径向渐变
    const gradient = this.ctx.createRadialGradient(
      mean.x, mean.y, 0, 
      mean.x, mean.y, radius
    );
    gradient.addColorStop(0, 'rgba(255,0,0,0.8)'); // 中心不透明红色
    gradient.addColorStop(1, 'rgba(255,0,0,0)');   // 边缘完全透明

    // 使用渐变填充区域绘制高斯分布效果
    this.ctx.fillStyle = gradient;
    this.ctx.fillRect(mean.x - radius, mean.y - radius, 2 * radius, 2 * radius);
  }
}

class HeatmapGenerator {
  constructor(canvas) {
    this.ctx = canvas.getContext('2d');      // 获取Canvas绘图上下文
    this.gmm = new GaussianMixture(5);      // 初始化5个聚类中心的高斯混合模型
  }

update(events) {
    const points = events.map(e => ({ x: e.x, y: e.y }));  // 提取坐标数据
    this.gmm.fit(points);                                  // 用事件点训练GMM模型

    this.ctx.clearRect(0, 0, canvas.width, canvas.height); // 清空画布
    this.gmm.components.forEach(cluster => {               // 遍历所有聚类
      this.drawGaussian(cluster.mean, cluster.cov);        // 绘制每个高斯分布
    });
  }

drawGaussian(mean, cov) {
    const radius = Math.sqrt(cov[0][0] + cov[1][1]) * 3;  // 计算分布半径（3倍标准差）
    const gradient = this.ctx.createRadialGradient(       // 创建径向渐变
      mean.x, mean.y, 0,              // 中心点
      mean.x, mean.y, radius          // 渐变半径
    );
    gradient.addColorStop(0, 'rgba(255,0,0,0.8)'); // 中心红色
    gradient.addColorStop(1, 'rgba(255,0,0,0)');   // 边缘透明

    this.ctx.fillStyle = gradient;
    this.ctx.fillRect(                             // 绘制渐变区域
      mean.x - radius, mean.y - radius,            // 起始坐标
      2 * radius, 2 * radius                       // 宽高
    );
  }