基于MongoDB的KOC项目实战分享
前段时间做的一个KOC项目，时间有点久远今天通过写文章的方式重新梳理一下。项目目标主要是用来统计KOL用户的粉丝数和文章的曝光阅读以及点赞收藏等数据的每日增长趋势。
实现方案
PHP+MongoDB
实现步骤
1、导入日志
目前日志系统分为服务端日志和客户端日志分为两个日志文件存储，本项目源数据是客户端日志文章曝光数据以及服务端日志里的文章阅读、粉丝取消关注数据。
这里有两个脚本，分别把客户端和服务端前一日的原始数据导入到mongodb中，这里遇到过导入速度效率不高的问题在下面会介绍。
2、分析mongodb中的原始数据
第一步导入日志处理完成后，会把mongo中的原始数据以用户维度和文章维度进行分别处理，统计出用户的粉丝和文章的数据增长趋势。
3、接口查询
根据用户ID查询mongo中处理后的数据返回
遇到的问题
1、导入日志时的插入问题
因为把操作mongo的各种方法是自己封装的，以便和PHP操作mysql的封装写法一致。封装mongo的时候只写了单条数据的插入，其实这种方式在没有索引大数据量插入的情况下效率也不会下降的太厉害，有索引特别是有多个索引效率就会看到明显差别，最后加入了批量插入。
每个索引占据一定的存储空间，在进行插入，更新和删除操作时也需要对索引进行操作。所以，如果你很少对集合进行读取操作，建议不使用索引。由于索引是存储在内存(RAM)中,你应该确保该索引的大小不超过内存的限制。如果索引的大小大于内存的限制，MongoDB会删除一些索引，这将导致性能下降。
和mysql一样，有些查询是不能被查询使用的，例如：正则表达式及非操作符，如 $nin, $not, 等；算术运算符，如 $mod等。
下面是数据不同方式插入的代码：
单条插入：
try {
$this->mgoBulk = new \MongoDB\Driver\BulkWrite;
$this->mgoBulk->insert($data);
$result = $this->mgoSDK->executeBulkWrite($this->db, $this->mgoBulk, $this->writeConcern);
return $result->getInsertedCount();
} catch (\MongoDB\Driver\Exception\BulkWriteException $e) {
return $e->getMessage();
}
批量插入：
try {
$this->mgoBulk = new \MongoDB\Driver\BulkWrite;
foreach ($data as $key => $val){
$this->mgoBulk->insert($val);
}
$result = $this->mgoSDK->executeBulkWrite($this->db, $this->mgoBulk, $this->writeConcern);
return $result->getInsertedCount();
} catch (\MongoDB\Driver\Exception\BulkWriteException $e) {
return $e->getMessage();
}
批量插入的优点是只用请求一次数据库。
2、查询
$mongolog = new \Library\MongoSDK($config, $initStatus, 'koc_log');
$result = $mongolog->select(['xwj_event' => 'delFollow','xwj_date' => $date], []);
封装的方法：/**
* @desc 查询
* @param array $query 查询条件
* @param array $fields 结果集返回的字段, array():表示返回所有字段 array('id'=>1, 'name'=>0):表示返回字段"id", 不返回字段"name"
* @param array $sort 排序字段, array('id'=>1):表示按id字段升序 array('id'=>-1):表示按id字段降序 array('id'=>1, 'age'=>-1):表示按id升序后再按age降序
* @param int $limit 取多少条记录
* @param int $skip 跳过多少条(从多少条开始) 可用于分页
* @return mixed[]
*/
public function select($query = [], $fields = [], $sort = [], $limit = 0, $skip = 0){
$options = [];
//指定返回哪些字段 1 表示返回 0表示不返回
if ($fields) {
$options['projection'] = $fields;
}
//指定排序字段
if ($sort) {
$options['sort'] = $sort;
}
//指定返回的条数
if($limit > 0){
$options['limit'] = $limit;
}
//指定起始位置
if($skip > 0){
$options['skip'] = $skip;
}
$query = new \MongoDB\Driver\Query($query, $options);
$cursor = $this->mgoSDK->executeQuery($this->db, $query);
$result = [];
foreach ($cursor as $val) {
$result[] = $this->_parseArr((array)$val);
}
return $result;
}
查询很简单和mysql查询基本类似。
3、聚合运算
部分例子：
$command = new \MongoDB\Driver\Command([
'aggregate' => 'apilogs',
'pipeline' => [
['$match' => ['add_date'=>['$gte' => '2020-04-28', '$lte' => '2020-05-12']]],
['$group' => ['_id' => ['api' => '$api', 'version' => '$version'], 'count' => ['$sum' => 1]]],
['$project' => ['version' => '$_id.version', 'api' => '$_id.api', 'count' => 1]],
['$sort' => ['count' => -1]],
],
'cursor' => new stdClass,
]);
try {
$cursor = $mgoSDK->executeCommand($mgodb, $command);
} catch(\MongoDB\Driver\Exception $e) {
//return [];
}
$data = [];
foreach ($cursor as $dv) {
$data[] = [
'api' => $dv->api,
'version' => $dv->version,
'count' => $dv->count,
];
}
这段代码是统计一段时间内API不同版本的调用数量。
较常用的关键字：
- $project：修改输入文档的结构。可以用来重命名、增加或删除域，也可以用于创建计算结果以及嵌套文档。
- $match使用MongoDB的标准查询操作。
- $limit：用来限制MongoDB聚合管道返回的文档数。
- $skip：在聚合管道中跳过指定数量的文档，并返回余下的文档。
- $group：将集合中的文档分组，可用于统计结果。
- $sort：将输入文档排序后输出。
mongodb虽然支持事务，但官方建议能不用就不用，所以，在你的项目中应用时，要注意这点。
但是mongodb提供了许多原子操作，比如文档的保存，修改，删除等，都是原子操作。所谓原子操作就是要么这个文档保存到Mongodb，要么没有保存到Mongodb，不会出现查询到的文档没有保存完整的情况。
About The Author
