当前位置：首页 > ds >正文

直播数据统计：如何让数据为我们所用？

ds 2025/7/16 6:43:08

文章目录

- 前言
一、数据日志接口
二、直播数据统计
三、观看行为统计
四、聊天词云
总结

前言

在线教育直播蓬勃发展的背后，海量数据蕴含着优化教学、提升运营的金钥匙。然而，原始数据如同沉睡的矿藏，数据日志接口是采集基础，直播数据统计描绘宏观效果，观看行为统计揭示个体学习轨迹，聊天词云则提炼课堂互动焦点。有效整合与分析这些数据，方能将其转化为精准决策的指南针，赋能教育者洞察学情、提升教学质量与用户粘性，真正让数据为我们所用。

一、数据日志接口

丰富API灵活调用，将数据后台与公司后台系统融合，帮助企业分析平台运营情况，提供学员的针对性教学及为市场精准营销。适合有技术能力的团队，对接接口可获取更详细原始数据，自定义分析维度，深度挖掘数据价值。
在这里插入图片描述
实现代码示例（Node.js + Express）

const express = require('express');
const app = express();
app.use(express.json());// 内存队列模拟（生产环境用Kafka/RabbitMQ）
const logQueue = [];// 1. 客户端埋点上报接口
app.post('/api/log', (req, res) => {const logData = {...req.body,timestamp: Date.now(),ip: req.ip};// 2. 写入缓冲队列logQueue.push(logData);// 立即响应避免阻塞客户端res.status(202).json({ status: 'queued' });
});// 3. 消费者处理（定时批量处理）
setInterval(() => {if(logQueue.length === 0) return;const batchSize = Math.min(100, logQueue.length);const batch = logQueue.splice(0, batchSize);// 4. 持久化存储（模拟数据库写入）console.log(`[DB] Inserting ${batchSize} logs:`);batch.forEach(log => {// 这里添加数据清洗逻辑const cleanLog = sanitizeLog(log);// 实际写入数据库操作（如InfluxDB）// db.insert('logs', cleanLog);});
}, 5000); // 每5秒处理一批// 数据清洗函数示例
function sanitizeLog(log) {return {event: String(log.event).substring(0, 50),userId: /^[a-z0-9_-]{1,36}$/i.test(log.userId) ? log.userId : 'invalid',duration: Math.max(0, parseInt(log.duration) || 0),// 添加教育场景特殊字段courseId: log.courseId || null,resourceType: ['video', 'ppt', 'quiz'].includes(log.resourceType)? log.resourceType : 'unknown'};
}// 教育场景专用事件上报
app.post('/api/edu/event', (req, res) => {const eduEvent = {eventType: 'edu_' + (req.body.eventType || 'custom'),...req.body,category: 'education'};logQueue.push(eduEvent);res.sendStatus(202);
});app.listen(3000, () => console.log('Logging API running on port 3000'));

客户端上报： 通过 POST /api/log 接收 JSON 日志数据

流量缓冲： 使用内存队列暂存日志（生产环境用消息队列）

批量处理： 定时任务批量获取日志（减少I/O压力）

数据清洗： sanitizeLog() 函数进行格式验证和安全处理

教育扩展： 专用/api/edu/event接口支持教学事件追踪

异步响应： 立即返回202状态码避免阻塞客户端

二、直播数据统计

直播后台提供直播数据统计，包括实时观看人数、历史观看人数、观看行为统计、观众信息统计等，帮助分析直播效果，推动市场决策。涵盖观看人数、点赞、礼物等核心数据。直播中控台，实时显示在线人数峰值、成交转化等，助力评估直播效果。
在这里插入图片描述
代码示例（Node.js + Redis）

const express = require('express');
const Redis = require('ioredis');
const app = express();
const redis = new Redis();// 实时数据上报接口
app.post('/api/live/stats', async (req, res) => {const { roomId, userId, event, value } = req.body;// 1. 实时计数器await redis.zincrby(`live:${roomId}:viewers`, 1, userId);if(event === 'gift') {await redis.zincrby(`live:${roomId}:gifts`, value, userId);}// 2. 实时弹幕处理if(event === 'comment') {redis.publish(`live-chat:${roomId}`, JSON.stringify({userId, text: value, timestamp: Date.now()}));}// 3. 观看时长统计if(event === 'leave') {const duration = Date.now() - parseInt(value);redis.hincrby(`live:${roomId}:duration`, userId, duration);}res.sendStatus(200);
});// 获取直播间实时数据
app.get('/api/live/stats/:roomId', async (req, res) => {const { roomId } = req.params;// 1. 获取在线观众数const onlineCount = await redis.zcard(`live:${roomId}:viewers`);// 2. 获取礼物榜TOP10const giftRank = await redis.zrevrange(`live:${roomId}:gifts`, 0, 9, 'WITHSCORES');// 3. 获取互动数据const interactions = {likes: await redis.get(`live:${roomId}:likes`),shares: await redis.get(`live:${roomId}:shares`)};res.json({ onlineCount, giftRank, interactions });
});// 教育场景专用统计
app.post('/api/edu/live/stats', async (req, res) => {const { roomId, courseId, event, userId } = req.body;// 课程相关统计if(event === 'resource_view') {await redis.hincrby(`course:${courseId}:resources`, req.body.resourceId, 1);}// 学习行为追踪if(event === 'question_answer') {redis.zincrby(`course:${courseId}:participation`, 1, userId);}res.sendStatus(200);
});// 启动服务
app.listen(4000, () => console.log('Live Stats API running'));

数据采集层
观看行为（进入/离开/时长），互动行为（点赞/评论/分享），付费行为（礼物/课程购买）

实时处理层
使用流处理引擎（如Flink）进行实时计算

存储层
实时数据：Redis（Sorted Sets存储排行榜）
历史数据：时序数据库（如InfluxDB）
分析数据：OLAP引擎（如ClickHouse）

可视化层
实时仪表盘：WebSocket推送
历史报表：定时任务生成

三、观看行为统计

使用POLYV的直播观看页，上传白名单到直播后台，即可统计每个观众的观看行为，包括何时进入频道、观看多长时间、是直播还是回看等等。POLYV还提供以上数据的导出，助您掌握每一位学员的学习进度。了解观众停留、互动等行为。像抖音直播后台，能看观众平均观看时长、转粉率，还能知道哪些时间段观众进出多，帮你优化节奏。
在这里插入图片描述
代码示例（Node.js + Redis + PostgreSQL）：

const express = require('express');
const Redis = require('ioredis');
const { Pool } = require('pg');
const app = express();
const redis = new Redis();
const pgPool = new Pool({ connectionString: 'postgres://user:pass@localhost/edu_stats' });// 观看行为上报接口
app.post('/api/watch/behavior', async (req, res) => {const { userId, videoId, event, timestamp, progress } = req.body;// 1. 实时行为记录（Redis）const watchKey = `watch:${videoId}:${userId}`;switch(event) {case 'start':// 记录开始时间await redis.hset(watchKey, 'start', timestamp);// 初始化进度数组await redis.rpush(`${watchKey}:progress`, progress);break;case 'progress':// 每10%记录一次进度if (progress % 10 === 0) {await redis.rpush(`${watchKey}:progress`, progress);}// 更新最后活跃时间await redis.hset(watchKey, 'last_active', timestamp);break;case 'complete':// 记录完整观看const duration = timestamp - await redis.hget(watchKey, 'start');await pgPool.query(`INSERT INTO watch_completions (user_id, video_id, duration) VALUES ($1, $2, $3)`, [userId, videoId, duration]);await redis.del(watchKey);break;}res.sendStatus(200);
});// 定时处理观看片段（每5分钟）
setInterval(async () => {// 2. 找出超时未活动的观看const keys = await redis.keys('watch:*:*');for (const key of keys) {const lastActive = await redis.hget(key, 'last_active');if (Date.now() - lastActive > 300000) { // 5分钟超时const [videoId, userId] = key.split(':').slice(1);const startTime = await redis.hget(key, 'start');const progress = await redis.lrange(`${key}:progress`, 0, -1);// 3. 保存观看片段到PostgreSQLawait pgPool.query(`INSERT INTO watch_segments (user_id, video_id, start_time, end_time, progress) VALUES ($1, $2, $3, $4, $5)`, [userId, videoId, startTime, lastActive, progress]);await redis.del(key, `${key}:progress`);}}
}, 300000);// 获取视频观看分析
app.get('/api/video/stats/:videoId', async (req, res) => {const { videoId } = req.params;// 4. 热度图计算const heatmap = await pgPool.query(`SELECT progress_bucket,COUNT(*) as view_countFROM (SELECT unnest(progress) / 10 * 10 AS progress_bucket FROM watch_segments WHERE video_id = $1) segmentsGROUP BY progress_bucketORDER BY progress_bucket`, [videoId]);// 5. 完播率计算const completionRate = await pgPool.query(`SELECT (SELECT COUNT(*) FROM watch_completions WHERE video_id = $1)::float /(SELECT COUNT(DISTINCT user_id) FROM watch_segments WHERE video_id = $1) AS completion_rate`, [videoId]);// 6. 平均观看时长const avgDuration = await pgPool.query(`SELECT AVG(EXTRACT(EPOCH FROM (end_time - start_time))) AS avg_duration FROM watch_segments WHERE video_id = $1`, [videoId]);res.json({videoId,heatmap: heatmap.rows,completionRate: completionRate.rows[0].completion_rate,avgDuration: avgDuration.rows[0].avg_duration});
});// 教育场景特殊分析
app.get('/api/edu/video/:courseId', async (req, res) => {const { courseId } = req.params;// 7. 知识点掌握分析const knowledgePoints = await pgPool.query(`SELECT kp.id,kp.title,COUNT(DISTINCT ws.user_id) FILTER (WHERE ws.progress @> ARRAY[kp.start_point]) AS viewed,COUNT(DISTINCT qa.user_id) FILTER (WHERE qa.score > 80) AS masteredFROM knowledge_points kpLEFT JOIN watch_segments ws ON kp.video_id = ws.video_idLEFT JOIN quiz_attempts qa ON kp.id = qa.knowledge_idWHERE kp.course_id = $1GROUP BY kp.id`, [courseId]);res.json(knowledgePoints.rows);
});app.listen(5000, () => console.log('Watch Analytics API running'));

四、聊天词云

聊天词云，即把聊天记录，做中文的分词，提取词频最高的50个，绘制出词云。分析观众聊天内容，了解观众对直播的看法等，助力挖掘聊天消息的最优业务价值。能直观展示观众高频讨论的话题。用微信指数，输入直播相关关键词，生成词云图，轻松 get 观众关注点，方便调整直播内容。
在这里插入图片描述

代码示例

const WebSocket = require('ws');
const redis = require('redis');
const nodejieba = require('nodejieba');
const client = redis.createClient();
const wss = new WebSocket.Server({ port: 8080 });// 配置停用词和权重规则
const STOP_WORDS = new Set(['的', '了', '和', '呢', '啊']);
const WEIGHT_RULES = {'名词': 1.5,'动词': 1.2,'形容词': 1.3
};// 1. 实时聊天处理
wss.on('connection', (ws) => {ws.on('message', (message) => {const { roomId, userId, text, timestamp } = JSON.parse(message);// 2. 基础消息存储client.lpush(`chat:${roomId}:raw`, text);client.zincrby(`chat:${roomId}:user`, 1, userId);// 3. 实时词频处理processText(roomId, text);// 4. 广播消息wss.clients.forEach(client => {if (client.roomId === roomId && client.readyState === WebSocket.OPEN) {client.send(JSON.stringify({ type: 'message', text, userId }));}});});// 加入房间处理ws.on('join', roomId => {ws.roomId = roomId;});
});// 5. 文本处理函数
async function processText(roomId, text) {// 中文分词const words = nodejieba.tag(text);// 词性过滤和权重计算words.forEach(({ word, tag }) => {if (!STOP_WORDS.has(word) && WEIGHT_RULES[tag]) {const weight = WEIGHT_RULES[tag] || 1;// 6. 词频统计（带权重）client.zincrby(`chat:${roomId}:words`, weight, word);}});
}// 7. 词云数据获取API
const express = require('express');
const app = express();
app.get('/api/wordcloud/:roomId', async (req, res) => {const { roomId } = req.params;const { limit = 50 } = req.query;// 8. 获取热词TOP Nconst wordData = await client.zrevrange(`chat:${roomId}:words`, 0, limit - 1, 'WITHSCORES');// 9. 格式化词云数据const wordCloud = [];for (let i = 0; i < wordData.length; i += 2) {wordCloud.push({text: wordData[i],value: parseFloat(wordData[i + 1])});}// 10. 获取高频用户（教育场景特别关注）const topUsers = await client.zrevrange(`chat:${roomId}:user`, 0, 4, 'WITHSCORES');res.json({ wordCloud, topUsers });
});// 11. 教育场景特殊处理
app.post('/api/edu/chat/keywords', async (req, res) => {const { courseId, keywords } = req.body;// 创建课程关键词监控keywords.forEach(keyword => {client.sadd(`course:${courseId}:keywords`, keyword);});// 启动关键词实时提醒const subscriber = redis.createClient();subscriber.subscribe(`course:${courseId}:alerts`);subscriber.on('message', (channel, message) => {wss.clients.forEach(client => {if (client.courseId === courseId) {client.send(JSON.stringify({ type: 'keyword_alert', keyword: message }));}});});res.sendStatus(200);
});// 12. 关键词触发检测
client.on('message', (channel, msg) => {const [_, roomId] = channel.split(':');const keywords = client.smembers(`course:${roomId}:keywords`);keywords.forEach(keyword => {if (msg.includes(keyword)) {client.publish(`course:${roomId}:alerts`, keyword);// 教育场景：记录关键词出现client.zincrby(`course:${roomId}:keyword_stats`, 1, keyword);}});
});// 启动服务
app.listen(3000, () => console.log('WordCloud API running'));
wss.on('listening', () => console.log('WebSocket Server running'));