/
main.go
74 lines (64 loc) · 1.44 KB
/
main.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
package main
import (
"database/sql"
"flag"
"log"
_ "github.com/chrislusf/glow/driver"
"github.com/chrislusf/glow/flow"
_ "github.com/lib/pq"
)
func iterate(dbURL string, fn func(rows *sql.Rows)) {
db, err := sql.Open("postgres", dbURL)
if err != nil {
log.Printf("error connection to the database: %s", err)
return
}
selectQuery := `select rowid, infohash, country, address from get_peers_log`
rows, err := db.Query(selectQuery) // select infohash from get_peers_log
if err != nil {
log.Println(err) // Any way to give feedback about errors?
return
}
fn(rows)
rows.Close()
}
const shards = 3
type result struct {
rowid string
infohash string
country sql.NullString
address string
}
var (
f = flow.New()
)
func init() {
f.Source(func(ch chan result) {
iterate("postgresql://maxroach@localhost:26257/roachy?sslmode=disable", func(rows *sql.Rows) {
for rows.Next() {
var r result
if err := rows.Scan(&r.rowid, &r.infohash, &r.country, &r.address); err != nil {
log.Println(err)
} else {
ch <- r
}
}
})
}, shards).Filter(func(row result) bool {
// find lines with empty Country
return !row.country.Valid // && row.country.String == ""
}).Map(func(row result) string {
return row.country.String
}).Map(func(key string) int {
return 1
}).Reduce(func(x int, y int) int {
return x + y
}).Map(func(x int) {
println("count:", x)
})
}
func main() {
flag.Parse()
flow.Ready()
f.Run()
}