通过日期范围动态计算项目数

【问题标题】：Count number of items dynamically through date range通过日期范围动态计算项目数
【发布时间】：2018-06-12 00:35:11
【问题描述】：

目前，我正在处理存储在 MongoDB 中的大量数据（20M 个更大的集合中的 2M 个单个集合）。 字段：id、项目名称、项目类型、项目描述和日期（）

动态计算整个集合在一周和一个月的日期范围内出现的项目数。即从 2014-01-01 到 2014-01-07 有 20 项，从 2014-01-08 到 2014-01-16 有 50 项，..等等

使用python，我怎样才能做到这一点？他们的库是为此还是自定义代码？

或者，这是否应该全部通过 MongoDB 完成？

【问题讨论】：

    标签： python mongodb mongodb-query aggregation-framework


    【解决方案1】：

    一般的方法当然是让数据库处理聚合。如果您想要“周范围”内的数据，那么有几种方法可以解决，具体取决于您实际需要哪种方法。

    按 ISO 周分组

    仅以“五月”为例进行演示，那么您将有类似的内容：

    startdate = datetime(2018,5,1)
enddate = datetime(2018,6,1)

result = db.sales.aggregate([
  { '$match': { 'date': { '$gte': startdate, '$lt': enddate } } },
  { '$group': {
    '_id': {
      'year': { '$year': '$date' },
      'week': { '$isoWeek': '$date' }
    },
    'totalQty': { '$sum': '$qty' },
    'count': { '$sum': 1 }
  }},
  { '$sort': { '_id': 1 } }
])

    这是一个相当简单的调用，使用 $year 和 $isoWeek 甚至可能是 $week 运算符，具体取决于您的 MongoDB 版本实际支持的内容。您需要做的就是在$group 的_id 分组键中指定这些，然后根据您实际需要在该分组中“累积”的内容选择其他累加器，例如$sum。

    $week 和 $isoWeek 只是略有不同，后者更符合 Python 的 isoweek 库的功能以及其他语言的类似功能。一般来说，您可以通过添加1 来调整这两个星期。有关详细信息，请参阅文档。

    在这种情况下，您可以选择让数据库进行“聚合”工作，然后根据输出获取“所需日期”。即对于python，您可以使用与每周对应的datetime 值将结果转换为：

    result = list(result)
for item in result:
  item.update({
    'start': datetime.combine(
      Week(item['_id']['year'],item['_id']['week']).monday(),
      datetime.min.time()
    ),
    'end': datetime.combine(
      Week(item['_id']['year'],item['_id']['week']).sunday(),
      datetime.max.time()
    )
  })
  item.pop('_id',None)

    按自定义分组

    如果您不适合坚持 ISO 标准，那么另一种方法是定义您自己的“间隔”，在该“间隔”上积累“分组”。 MongoDB这里的主要工具是$bucket，并预先进行一点列表处理：

    cuts = [startdate]
date = startdate

while ( date < enddate ):
  date = date + timedelta(days=7)
  if ( date > enddate ):
    date = enddate
  cuts.append(date)

alternate = db.sales.aggregate([
  { '$match': { 'date': { '$gte': startdate, '$lt': enddate } } },
  { '$bucket': {
    'groupBy': '$date',
    'boundaries': cuts,
    'output': {
      'totalQty': { '$sum': '$qty' },
      'count': { '$sum': 1 }
    }
  }},
  { '$project': {
    '_id': 0,
    'start': '$_id',
    'end': {
      '$cond': {
        'if': {
          '$gt': [
            { '$add': ['$_id', (1000 * 60 * 60 * 24 * 7) - 1] },
            enddate
          ]
        },
        'then': { '$add': [ enddate, -1 ] },
        'else': {
          '$add': ['$_id', (1000 * 60 * 60 * 24 * 7) - 1]
        }
      }
    },
    'totalQty': 1,
    'count': 1
  }}
])

    我们不使用 $week 或 $isoWeek 等定义的函数，而是从给定的查询开始日期计算出“7 天的间隔”并生成这些间隔的数组，当然总是以“所选数据范围中的最大值”值。

    这个list 然后在$bucket 聚合阶段的参数中给出它的"boundaries" 选项。这实际上只是一个值列表，它告诉语句为每个产生的“分组”累积“最多”什么。

    实际的语句实际上只是$switch 聚合运算符在$group 管道阶段中的“速记”实现。这两个运算符都需要 MongoDB 3.4，但实际上您可以在 $group 中使用 $cond 执行相同的操作，但只需为每个“边界”值嵌套每个 else 条件。这是可能的，但只是涉及更多一点，你现在真的应该使用 MongoDB 3.4 作为最低版本。

    如果你发现你真的必须这样做，在$group 中使用$cond 被添加到下面的示例中，只是展示了如何从本质上将相同的cuts 列表转换为这样的语句，这意味着你基本上可以做到一直到引入聚合框架的 MongoDB 2.2 也是如此。

    示例

    作为一个完整示例，您可以考虑以下清单，该清单插入一个月的随机数据，然后在其上运行两个呈现的聚合选项：

    from random import randint
from datetime import datetime, timedelta, date
from isoweek import Week

from pymongo import MongoClient
from bson.json_util import dumps, JSONOptions
import bson.json_util

client = MongoClient()
db = client.test

db.sales.delete_many({})

startdate = datetime(2018,5,1)
enddate = datetime(2018,6,1)

currdate = startdate

batch = []

while ( currdate < enddate ):
  currdate = currdate + timedelta(hours=randint(1,24))
  if ( currdate > enddate ):
    currdate = enddate
  qty = randint(1,100);
  if ( currdate < enddate ):
    batch.append({ 'date': currdate, 'qty': qty })

  if ( len(batch) >= 1000 ):
    db.sales.insert_many(batch)
    batch = []

if ( len(batch) > 0):
  db.sales.insert_many(batch)
  batch = []

result = db.sales.aggregate([
  { '$match': { 'date': { '$gte': startdate, '$lt': enddate } } },
  { '$group': {
    '_id': {
      'year': { '$year': '$date' },
      'week': { '$isoWeek': '$date' }
    },
    'totalQty': { '$sum': '$qty' },
    'count': { '$sum': 1 }
  }},
  { '$sort': { '_id': 1 } }
])

result = list(result)
for item in result:
  item.update({
    'start': datetime.combine(
      Week(item['_id']['year'],item['_id']['week']).monday(),
      datetime.min.time()
    ),
    'end': datetime.combine(
      Week(item['_id']['year'],item['_id']['week']).sunday(),
      datetime.max.time()
    )
  })
  item.pop('_id',None)

print("Week grouping")
print(
  dumps(result,indent=2,
    json_options=JSONOptions(datetime_representation=2)))

cuts = [startdate]
date = startdate

while ( date < enddate ):
  date = date + timedelta(days=7)
  if ( date > enddate ):
    date = enddate
  cuts.append(date)

alternate = db.sales.aggregate([
  { '$match': { 'date': { '$gte': startdate, '$lt': enddate } } },
  { '$bucket': {
    'groupBy': '$date',
    'boundaries': cuts,
    'output': {
      'totalQty': { '$sum': '$qty' },
      'count': { '$sum': 1 }
    }
  }},
  { '$project': {
    '_id': 0,
    'start': '$_id',
    'end': {
      '$cond': {
        'if': {
          '$gt': [
            { '$add': ['$_id', (1000 * 60 * 60 * 24 * 7) - 1] },
            enddate
          ]
        },
        'then': { '$add': [ enddate, -1 ] },
        'else': {
          '$add': ['$_id', (1000 * 60 * 60 * 24 * 7) - 1]
        }
      }
    },
    'totalQty': 1,
    'count': 1
  }}
])

alternate = list(alternate)

print("Bucket grouping")
print(
  dumps(alternate,indent=2,
    json_options=JSONOptions(datetime_representation=2)))

cuts = [startdate]
date = startdate

while ( date < enddate ):
  date = date + timedelta(days=7)
  if ( date > enddate ):
    date = enddate
  if ( date < enddate ):
    cuts.append(date)

stack = []

for i in range(len(cuts)-1,0,-1):
  rec = {
    '$cond': [
      { '$lt': [ '$date', cuts[i] ] },
      cuts[i-1]
    ]
  }

  if ( len(stack) == 0 ):
    rec['$cond'].append(cuts[i])
  else:
    lval = stack.pop()
    rec['$cond'].append(lval)

  stack.append(rec)

pipeline = [
  { '$match': { 'date': { '$gt': startdate, '$lt': enddate } } },
  { '$group': {
    '_id': stack[0],
    'totalQty': { '$sum': '$qty' },
    'count': { '$sum': 1 }
  }},
  { '$sort': { '_id': 1 } },
  { '$project': {
    '_id': 0,
    'start': '$_id',
    'end': {
      '$cond': {
        'if': {
          '$gt': [
            { '$add': [ '$_id', ( 1000 * 60 * 60 * 24 * 7 ) - 1 ] },
            enddate
          ]
        },
        'then': { '$add': [ enddate, -1 ] },
        'else': {
          '$add': [ '$_id', ( 1000 * 60 * 60 * 24 * 7 ) - 1 ]
        }
      }
    },
    'totalQty': 1,
    'count': 1
  }}
]

#print(
#  dumps(pipeline,indent=2,
#    json_options=JSONOptions(datetime_representation=2)))

older = db.sales.aggregate(pipeline)
older = list(older)

print("Cond Group")
print(
  dumps(older,indent=2,
    json_options=JSONOptions(datetime_representation=2)))

    有输出：

    Week grouping
[
  {
    "totalQty": 449,
    "count": 9,
    "start": {
      "$date": "2018-04-30T00:00:00Z"
    },
    "end": {
      "$date": "2018-05-06T23:59:59.999Z"
    }
  },
  {
    "totalQty": 734,
    "count": 14,
    "start": {
      "$date": "2018-05-07T00:00:00Z"
    },
    "end": {
      "$date": "2018-05-13T23:59:59.999Z"
    }
  },
  {
    "totalQty": 686,
    "count": 14,
    "start": {
      "$date": "2018-05-14T00:00:00Z"
    },
    "end": {
      "$date": "2018-05-20T23:59:59.999Z"
    }
  },
  {
    "totalQty": 592,
    "count": 12,
    "start": {
      "$date": "2018-05-21T00:00:00Z"
    },
    "end": {
      "$date": "2018-05-27T23:59:59.999Z"
    }
  },
  {
    "totalQty": 205,
    "count": 6,
    "start": {
      "$date": "2018-05-28T00:00:00Z"
    },
    "end": {
      "$date": "2018-06-03T23:59:59.999Z"
    }
  }
]
Bucket grouping
[
  {
    "totalQty": 489,
    "count": 11,
    "start": {
      "$date": "2018-05-01T00:00:00Z"
    },
    "end": {
      "$date": "2018-05-07T23:59:59.999Z"
    }
  },
  {
    "totalQty": 751,
    "count": 13,
    "start": {
      "$date": "2018-05-08T00:00:00Z"
    },
    "end": {
      "$date": "2018-05-14T23:59:59.999Z"
    }
  },
  {
    "totalQty": 750,
    "count": 15,
    "start": {
      "$date": "2018-05-15T00:00:00Z"
    },
    "end": {
      "$date": "2018-05-21T23:59:59.999Z"
    }
  },
  {
    "totalQty": 493,
    "count": 11,
    "start": {
      "$date": "2018-05-22T00:00:00Z"
    },
    "end": {
      "$date": "2018-05-28T23:59:59.999Z"
    }
  },
  {
    "totalQty": 183,
    "count": 5,
    "start": {
      "$date": "2018-05-29T00:00:00Z"
    },
    "end": {
      "$date": "2018-05-31T23:59:59.999Z"
    }
  }
]
Cond Group
[
  {
    "totalQty": 489,
    "count": 11,
    "start": {
      "$date": "2018-05-01T00:00:00Z"
    },
    "end": {
      "$date": "2018-05-07T23:59:59.999Z"
    }
  },
  {
    "totalQty": 751,
    "count": 13,
    "start": {
      "$date": "2018-05-08T00:00:00Z"
    },
    "end": {
      "$date": "2018-05-14T23:59:59.999Z"
    }
  },
  {
    "totalQty": 750,
    "count": 15,
    "start": {
      "$date": "2018-05-15T00:00:00Z"
    },
    "end": {
      "$date": "2018-05-21T23:59:59.999Z"
    }
  },
  {
    "totalQty": 493,
    "count": 11,
    "start": {
      "$date": "2018-05-22T00:00:00Z"
    },
    "end": {
      "$date": "2018-05-28T23:59:59.999Z"
    }
  },
  {
    "totalQty": 183,
    "count": 5,
    "start": {
      "$date": "2018-05-29T00:00:00Z"
    },
    "end": {
      "$date": "2018-05-31T23:59:59.999Z"
    }
  }
]

    可选的 JavaScript 演示

    由于上面的一些方法相当“pythonic”，那么对于更广泛的 JavaScript 大脑受众来说，该主题的共同点将是：

    const { Schema } = mongoose = require('mongoose');
const moment = require('moment');

const uri = 'mongodb://localhost/test';

mongoose.Promise = global.Promise;
//mongoose.set('debug',true);

const saleSchema = new Schema({
  date: Date,
  qty: Number
})

const Sale = mongoose.model('Sale', saleSchema);

const log = data => console.log(JSON.stringify(data, undefined, 2));

(async function() {

  try {

    const conn = await mongoose.connect(uri);

    let start = new Date("2018-05-01");
    let end = new Date("2018-06-01");
    let date = new Date(start.valueOf());

    await Promise.all(Object.entries(conn.models).map(([k,m]) => m.remove()));

    let batch = [];

    while ( date.valueOf() < end.valueOf() ) {
      let hour = Math.floor(Math.random() * 24) + 1;
      date = new Date(date.valueOf() + (1000 * 60 * 60 * hour));
      if ( date > end )
        date = end;
      let qty = Math.floor(Math.random() * 100) + 1;
      if (date < end)
        batch.push({ date, qty });

      if (batch.length >= 1000) {
        await Sale.insertMany(batch);
        batch = [];
      }
    }

    if (batch.length > 0) {
      await Sale.insertMany(batch);
      batch = [];
    }

    let result = await Sale.aggregate([
      { "$match": { "date": { "$gte": start, "$lt": end } } },
      { "$group": {
        "_id": {
          "year": { "$year": "$date" },
          "week": { "$isoWeek": "$date" }
        },
        "totalQty": { "$sum": "$qty" },
        "count": { "$sum": 1 }
      }},
      { "$sort": { "_id": 1 } }
    ]);

    result = result.map(({ _id: { year, week }, ...r }) =>
      ({
        start: moment.utc([year]).isoWeek(week).startOf('isoWeek').toDate(),
        end: moment.utc([year]).isoWeek(week).endOf('isoWeek').toDate(),
        ...r
      })
    );

    log({ name: 'ISO group', result });

    let cuts = [start];
    date = start;

    while ( date.valueOf() < end.valueOf() ) {
      date = new Date(date.valueOf() + ( 1000 * 60 * 60 * 24 * 7 ));
      if ( date.valueOf() > end.valueOf() ) date = end;
      cuts.push(date);
    }

    let alternate = await Sale.aggregate([
      { "$match": { "date": { "$gte": start, "$lt": end } } },
      { "$bucket": {
        "groupBy": "$date",
        "boundaries": cuts,
        "output": {
          "totalQty": { "$sum": "$qty" },
          "count": { "$sum": 1 }
        }
      }},
      { "$addFields": {
        "_id": "$$REMOVE",
        "start": "$_id",
        "end": {
          "$cond": {
            "if": {
              "$gt": [
                { "$add": [ "$_id", ( 1000 * 60 * 60 * 24 * 7 ) - 1 ] },
                end
              ]
            },
            "then": { "$add": [ end, -1 ] },
            "else": {
              "$add": [ "$_id", ( 1000 * 60 * 60 * 24 * 7 ) - 1 ]
            }
          }
        }
      }}
    ]);
    log({ name: "Bucket group", result: alternate });


    cuts = [start];
    date = start;

    while ( date.valueOf() < end.valueOf() ) {
      date = new Date(date.valueOf() + ( 1000 * 60 * 60 * 24 * 7 ));
      if ( date.valueOf() > end.valueOf() ) date = end;
      if ( date.valueOf() < end.valueOf() )
        cuts.push(date);
    }

    let stack = [];

    for ( let i = cuts.length - 1; i > 0; i-- ) {
      let rec = {
        "$cond": [
          { "$lt": [ "$date", cuts[i] ] },
          cuts[i-1]
        ]
      };

      if ( stack.length === 0 ) {
        rec['$cond'].push(cuts[i])
      } else {
        let lval = stack.pop();
        rec['$cond'].push(lval);
      }

      stack.push(rec);
    }

    let pipeline = [
      { "$group": {
        "_id": stack[0],
        "totalQty": { "$sum": "$qty" },
        "count": { "$sum": 1 }
      }},
      { "$sort": { "_id": 1 } },
      { "$project": {
        "_id": 0,
        "start": "$_id",
        "end": {
          "$cond": {
            "if": {
              "$gt": [
                { "$add": [ "$_id", ( 1000 * 60 * 60 * 24 * 7 ) - 1 ] },
                end
              ]
            },
            "then": { "$add": [ end, -1 ] },
            "else": {
              "$add": [ "$_id", ( 1000 * 60 * 60 * 24 * 7 ) - 1 ]
            }
          }
        },
        "totalQty": 1,
        "count": 1
      }}
    ];

    let older = await Sale.aggregate(pipeline);
    log({ name: "Cond group", result: older });

    mongoose.disconnect();

  } catch(e) {
    console.error(e)
  } finally {
    process.exit()
  }

})()

    当然还有类似的输出：

    {
  "name": "ISO group",
  "result": [
    {
      "start": "2018-04-30T00:00:00.000Z",
      "end": "2018-05-06T23:59:59.999Z",
      "totalQty": 576,
      "count": 10
    },
    {
      "start": "2018-05-07T00:00:00.000Z",
      "end": "2018-05-13T23:59:59.999Z",
      "totalQty": 707,
      "count": 11
    },
    {
      "start": "2018-05-14T00:00:00.000Z",
      "end": "2018-05-20T23:59:59.999Z",
      "totalQty": 656,
      "count": 12
    },
    {
      "start": "2018-05-21T00:00:00.000Z",
      "end": "2018-05-27T23:59:59.999Z",
      "totalQty": 829,
      "count": 16
    },
    {
      "start": "2018-05-28T00:00:00.000Z",
      "end": "2018-06-03T23:59:59.999Z",
      "totalQty": 239,
      "count": 6
    }
  ]
}
{
  "name": "Bucket group",
  "result": [
    {
      "totalQty": 666,
      "count": 11,
      "start": "2018-05-01T00:00:00.000Z",
      "end": "2018-05-07T23:59:59.999Z"
    },
    {
      "totalQty": 727,
      "count": 12,
      "start": "2018-05-08T00:00:00.000Z",
      "end": "2018-05-14T23:59:59.999Z"
    },
    {
      "totalQty": 647,
      "count": 12,
      "start": "2018-05-15T00:00:00.000Z",
      "end": "2018-05-21T23:59:59.999Z"
    },
    {
      "totalQty": 743,
      "count": 15,
      "start": "2018-05-22T00:00:00.000Z",
      "end": "2018-05-28T23:59:59.999Z"
    },
    {
      "totalQty": 224,
      "count": 5,
      "start": "2018-05-29T00:00:00.000Z",
      "end": "2018-05-31T23:59:59.999Z"
    }
  ]
}
{
  "name": "Cond group",
  "result": [
    {
      "totalQty": 666,
      "count": 11,
      "start": "2018-05-01T00:00:00.000Z",
      "end": "2018-05-07T23:59:59.999Z"
    },
    {
      "totalQty": 727,
      "count": 12,
      "start": "2018-05-08T00:00:00.000Z",
      "end": "2018-05-14T23:59:59.999Z"
    },
    {
      "totalQty": 647,
      "count": 12,
      "start": "2018-05-15T00:00:00.000Z",
      "end": "2018-05-21T23:59:59.999Z"
    },
    {
      "totalQty": 743,
      "count": 15,
      "start": "2018-05-22T00:00:00.000Z",
      "end": "2018-05-28T23:59:59.999Z"
    },
    {
      "totalQty": 224,
      "count": 5,
      "start": "2018-05-29T00:00:00.000Z",
      "end": "2018-05-31T23:59:59.999Z"
    }
  ]
}

    【讨论】：

    • 感谢您非常详细的解释和回答。我需要一点时间来消化和理解它。我将实施解决方案并回复您。
    • @Jin 如果您发现内容有用，那么您应该"upvote"，因为您的新权限现在允许。作为一个相对较新的用户帐户，当您接受答案时，通常会提示您这样做，但新 UI 中可能已经实现了更改，否则您可能通常不知道其重要性。另请参阅how to vote
    • @Jin 我认为您在投赞成票时不小心取消了“接受”。除非出于某种原因不再接受答案？
    • 可能写错了，我马上更新。感谢您的澄清。另外，我在一个单独的线程中还有一个问题，你能看一下吗？ stackoverflow.com/questions/50842664/…
    猜你喜欢
    • 2023-03-06
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    • 2020-01-13
    • 1970-01-01
    • 2011-02-05
    • 2017-05-16
    • 1970-01-01
    相关资源
    最近更新 更多
    热门标签
    Java Python linux javascript Mysql C# Docker 算法 前端 SpringBoot Redis Vue spring 设计模式 .net core .net kubernetes c++ 数据库 数据结构 大数据 js 机器学习 微服务 Android Go 程序员 面试 JVM ASP.net core 云原生 人工智能 后端 PHP git CSS golang k8s Nginx Django mybatis 深度学习 多线程 React 架构 devops 爬虫 云计算 Spring Boot LeetCode