This repository has been archived on 2025-03-01. You can view files and clone it, but cannot push or open issues or pull requests.
trantor/mapreduce.go
2014-02-18 00:47:58 +01:00

258 lines
8 KiB
Go

package main
import (
"labix.org/v2/mgo"
"labix.org/v2/mgo/bson"
"time"
)
func GetTags(numTags int, tagsColl *mgo.Collection) ([]string, error) {
var result []struct {
Tag string "_id"
}
err := tagsColl.Find(nil).Sort("-value").Limit(numTags).All(&result)
if err != nil {
return nil, err
}
tags := make([]string, len(result))
for i, r := range result {
tags[i] = r.Tag
}
return tags, nil
}
func GetBooksVisited(num int, visitedColl *mgo.Collection) ([]bson.ObjectId, error) {
var result []struct {
Book bson.ObjectId "_id"
}
err := visitedColl.Find(nil).Sort("-value").Limit(num).All(&result)
if err != nil {
return nil, err
}
books := make([]bson.ObjectId, len(result))
for i, r := range result {
books[i] = r.Book
}
return books, nil
}
func GetVisits(visitsColl *mgo.Collection) ([]Visits, error) {
var result []Visits
err := visitsColl.Find(nil).All(&result)
return result, err
}
type MR struct {
database *mgo.Database
}
func NewMR(database *mgo.Database) *MR {
m := new(MR)
m.database = database
return m
}
func (m *MR) UpdateTags(booksColl *mgo.Collection) error {
var mr mgo.MapReduce
mr.Map = `function() {
if (this.subject) {
this.subject.forEach(function(s) { emit(s, 1); });
}
}`
mr.Reduce = `function(tag, vals) {
var count = 0;
vals.forEach(function() { count += 1; });
return count;
}`
return m.update(&mr, bson.M{"active": true}, booksColl, TAGS_COLL)
}
func (m *MR) UpdateMostVisited(statsColl *mgo.Collection) error {
return m.updateMostBooks(statsColl, "book", VISITED_COLL)
}
func (m *MR) UpdateMostDownloaded(statsColl *mgo.Collection) error {
return m.updateMostBooks(statsColl, "download", DOWNLOADED_COLL)
}
func (m *MR) updateMostBooks(statsColl *mgo.Collection, section string, resColl string) error {
const numDays = 30
start := time.Now().UTC().Add(-numDays * 24 * time.Hour)
var mr mgo.MapReduce
mr.Map = `function() {
emit(this.id, 1);
}`
mr.Reduce = `function(tag, vals) {
var count = 0;
vals.forEach(function() { count += 1; });
return count;
}`
return m.update(&mr, bson.M{"date": bson.M{"$gt": start}, "section": section}, statsColl, resColl)
}
func (m *MR) UpdateHourVisits(statsColl *mgo.Collection) error {
const numDays = 2
start := time.Now().UTC().Add(-numDays * 24 * time.Hour)
const reduce = `function(date, vals) {
var count = 0;
vals.forEach(function(v) { count += v; });
return count;
}`
var mr mgo.MapReduce
mr.Map = `function() {
var date = Date.UTC(this.date.getUTCFullYear(),
this.date.getUTCMonth(),
this.date.getUTCDate(),
this.date.getUTCHours());
emit({date: date, session: this.session}, 1);
}`
mr.Reduce = reduce
err := m.update(&mr, bson.M{"date": bson.M{"$gte": start}}, statsColl, HOURLY_VISITS_COLL+"_raw")
if err != nil {
return err
}
var mr2 mgo.MapReduce
mr2.Map = `function() {
emit(this['_id']['date'], 1);
}`
mr2.Reduce = reduce
hourly_raw := m.database.C(HOURLY_VISITS_COLL + "_raw")
return m.update(&mr2, bson.M{}, hourly_raw, HOURLY_VISITS_COLL)
}
func (m *MR) UpdateDayVisits(statsColl *mgo.Collection) error {
const numDays = 30
start := time.Now().UTC().Add(-numDays * 24 * time.Hour).Truncate(24 * time.Hour)
const reduce = `function(date, vals) {
var count = 0;
vals.forEach(function(v) { count += v; });
return count;
}`
var mr mgo.MapReduce
mr.Map = `function() {
var date = Date.UTC(this.date.getUTCFullYear(),
this.date.getUTCMonth(),
this.date.getUTCDate());
emit({date: date, session: this.session}, 1);
}`
mr.Reduce = reduce
err := m.update(&mr, bson.M{"date": bson.M{"$gte": start}}, statsColl, DAILY_VISITS_COLL+"_raw")
if err != nil {
return err
}
var mr2 mgo.MapReduce
mr2.Map = `function() {
emit(this['_id']['date'], 1);
}`
mr2.Reduce = reduce
daily_raw := m.database.C(DAILY_VISITS_COLL + "_raw")
return m.update(&mr2, bson.M{}, daily_raw, DAILY_VISITS_COLL)
}
func (m *MR) UpdateMonthVisits(statsColl *mgo.Collection) error {
const numDays = 365
start := time.Now().UTC().Add(-numDays * 24 * time.Hour).Truncate(24 * time.Hour)
const reduce = `function(date, vals) {
var count = 0;
vals.forEach(function(v) { count += v; });
return count;
}`
var mr mgo.MapReduce
mr.Map = `function() {
var date = Date.UTC(this.date.getUTCFullYear(),
this.date.getUTCMonth());
emit({date: date, session: this.session}, 1);
}`
mr.Reduce = reduce
err := m.update(&mr, bson.M{"date": bson.M{"$gte": start}}, statsColl, MONTHLY_VISITS_COLL+"_raw")
if err != nil {
return err
}
var mr2 mgo.MapReduce
mr2.Map = `function() {
emit(this['_id']['date'], 1);
}`
mr2.Reduce = reduce
monthly_raw := m.database.C(MONTHLY_VISITS_COLL + "_raw")
return m.update(&mr2, bson.M{}, monthly_raw, MONTHLY_VISITS_COLL)
}
func (m *MR) UpdateHourDownloads(statsColl *mgo.Collection) error {
const numDays = 2
start := time.Now().UTC().Add(-numDays * 24 * time.Hour)
var mr mgo.MapReduce
mr.Map = `function() {
var date = Date.UTC(this.date.getUTCFullYear(),
this.date.getUTCMonth(),
this.date.getUTCDate(),
this.date.getUTCHours());
emit(date, 1);
}`
mr.Reduce = `function(date, vals) {
var count = 0;
vals.forEach(function(v) { count += v; });
return count;
}`
return m.update(&mr, bson.M{"date": bson.M{"$gte": start}, "section": "download"}, statsColl, HOURLY_DOWNLOADS_COLL)
}
func (m *MR) UpdateDayDownloads(statsColl *mgo.Collection) error {
const numDays = 30
start := time.Now().UTC().Add(-numDays * 24 * time.Hour).Truncate(24 * time.Hour)
var mr mgo.MapReduce
mr.Map = `function() {
var date = Date.UTC(this.date.getUTCFullYear(),
this.date.getUTCMonth(),
this.date.getUTCDate());
emit(date, 1);
}`
mr.Reduce = `function(date, vals) {
var count = 0;
vals.forEach(function(v) { count += v; });
return count;
}`
return m.update(&mr, bson.M{"date": bson.M{"$gte": start}, "section": "download"}, statsColl, DAILY_DOWNLOADS_COLL)
}
func (m *MR) UpdateMonthDownloads(statsColl *mgo.Collection) error {
const numDays = 365
start := time.Now().UTC().Add(-numDays * 24 * time.Hour).Truncate(24 * time.Hour)
var mr mgo.MapReduce
mr.Map = `function() {
var date = Date.UTC(this.date.getUTCFullYear(),
this.date.getUTCMonth());
emit(date, 1);
}`
mr.Reduce = `function(date, vals) {
var count = 0;
vals.forEach(function(v) { count += v; });
return count;
}`
return m.update(&mr, bson.M{"date": bson.M{"$gte": start}, "section": "download"}, statsColl, MONTHLY_DOWNLOADS_COLL)
}
func (m *MR) update(mr *mgo.MapReduce, query bson.M, queryColl *mgo.Collection, storeColl string) error {
metaColl := m.database.C(META_COLL)
_, err := metaColl.RemoveAll(bson.M{"type": storeColl})
if err != nil {
return err
}
mr.Out = bson.M{"replace": storeColl}
_, err = queryColl.Find(query).MapReduce(mr, nil)
if err != nil {
return err
}
return metaColl.Insert(bson.M{"type": storeColl})
}