MongoDB千万级数据的分析

MongoDB千万级数据的分析 一、导入 清单1： 读取CSV文件，存储到数据库中 01 #-*- coding:UTF-8 -*- 02 03 Created on 2013-10-20 04 05 06 @author: tyk 07 08 09 10 11 from pymongo.connection import Connection 12 from time import time 13 import code

MongoDB千万级数据的分析

 

一、导入

 

清单1：

 

读取CSV文件，存储到数据库中

 

01

#-*- coding:UTF-8 -*-

02

'''

03

Created on 2013-10-20

04

 

05

 

06

@author: tyk

07

 

08

 

09

 

10

'''

11

from pymongo.connection import Connection

12

from time import time

13

import codecs

14

import csv

15

import os

16

rootdir = "2000W/"   # 指明被遍历的文件夹

17

'''

18

 

19

'''

20

def process_data():

21

  conn = Connection('localhost', 27017) #获取一个连接

22

  ##conn.drop_database('guestHouse') 

23

  db = conn.TYK 

24

  guest = db.guestHouse  

25

 

26

 

27

  guest_info = []

28

  for parent, dirnames, filenames in os.walk(rootdir):  #三个参数：分别返回1.父目录 2.所有文件夹名字（不含路径） 3.所有文件名字

29

  for filename in filenames:  

30

  ErrorLine = []

31

  key_length = 0

32

  fullname = os.path.join(parent,filename)

33

  try:

34

  #with codecs.open(fullname, encoding='utf_8') as file:

35

  with codecs.open(fullname, encoding='utf_8_sig') as file:#忽略UTF-8文件前面的BOM

36

  keys = file.readline().split(',')#先读掉第一行的注释

37

  key_length = len(keys)

38

  spamreader = csv.reader(file)#以CSV格式读取，返回的不再是str,而是list

39

  for line in spamreader:

40

  if key_length != len(line):#部分数据不完整，记录下来

41

  ErrorLine.append(line)

42

  else:

43

  each_info = {}

44

  for i in range(1, len(keys)):#过滤第一个字段Name,姓名将不再存到数据库中

45

  each_info[keys[i]] = line[i]

46

 

47

  guest_info.append(each_info)

48

  if len(guest_info) == 10000:#每10000条进行一次存储操作

49

  guest.insert(guest_info)  

50

  guest_info = []

51

 

52

  except Exception, e:

53

  print filename + "\t" + str(e)

54

 

55

  #统一处理错误信息

56

  with open('ERR/' + os.path.splitext(filename)[0] + '-ERR.csv', 'w') as log_file:

57

  spamwriter = csv.writer(log_file)

58

  for line in ErrorLine:

59

  spamwriter.writerow(line)

60

  #最后一批

61

  guest.insert(guest_info)  

62

 

63

if __name__ == '__main__':

64

  start = time()

65

  process_data()

66

  stop = time()

67

  print(str(stop-start) + "秒")

后来睡着了、关机了，耗时多久也不得而知了⊙﹏⊙b汗

 

总结：

 

1.文件编码为UTF-8，不能直接open()打开读取。

 

2.文件已CSV格式进行存储，读取时用CSV模块处理来读取。这是读出来的数据每行为一个list。注意，不能简单的以","拆分后进行读取。对于这种形状"a,b,c", d的数据是无法正确解析的。

 

3.对于UTF-8文件，如果有BOM的形式去读是要以'utf_8_sig'编码读取，这样会跳过开头的BOM。如果不处理掉BOM，BOM会随数据一同存到数据库中,造成类似" XXX"的现象（有一个空格的假象）。

 

如果真的已经存到库中了，那只有改key了

 

1

db.guestHouse.update({}, {"$rename" : {" Name" : "Name"}}, false, true)

另外，网上还有一种方法（尝试失败了，具体原因应该是把字符串转换成字节码然后再去比较。怎么转这个我还不会...）

 

1

#with codecs.open(fullname, encoding='utf-8') as file:

2

with codecs.open(fullname, encoding='utf_8_sig') as file:

3

  keys = file.readline().split(',')

4

  if keys[0][:3] == codecs.BOM_UTF8:#将keys[0]转化为字节码再去比较

5

  keys[0] = keys[0][3:]

扩展：

 

今天发现MongoDB本身就带有导入功能mongoimport，可以直接导入CSV文件...

 

小试一把

 

1.不做错误数据过滤，直接导入。用专利引用数据做一下实验（《Hadoop权威指南》一书中的实验数据）

 

实验数据：

 

01

"PATENT","GYEAR","GDATE","APPYEAR","COUNTRY","POSTATE","ASSIGNEE","ASSCODE","CLAIMS","NCLASS","CAT","SUBCAT","CMADE","CRECEIVE","RATIOCIT","GENERAL","ORIGINAL","FWDAPLAG","BCKGTLAG","SELFCTUB","SELFCTLB","SECDUPBD","SECDLWBD"

02

3070801,1963,1096,,"BE","",,1,,269,6,69,,1,,0,,,,,,,

03

3070802,1963,1096,,"US","TX",,1,,2,6,63,,0,,,,,,,,,

04

3070803,1963,1096,,"US",

05

"IL",,1,,2,6,63,,9,,0.3704,,,,,,,

06

3070804,1963,1096,,"US","OH",,1,,2,6,63,,3,,0.6667,,,,,,,

07

3070805,1963,1096,,"US","CA",,1,,2,6,63,,1,,0,,,,,,,

08

3070806,1963,1096,,"US","PA",,1,,2,6,63,,0,,,,,,,,,

09

3070807,1963,1096,,"US","OH",,1,,623,3,39,,3,,0.4444,,,,,,,

10

3070808,1963,1096,,"US","IA",,1,,623,3,39,,4,,0.375,,,,,,,

11

3070809,1963,1096,,,,1,,4,6,65,,0,,,,,,,,,

1

mongoimport -d TYK -c guest --type csv --file d:\text.csv --headerline

一共11行。第一行注释，9条数据。第3条中间截断，第9条取出中间两个数值"US","AZ"。按照csv规定现在应该是10条数据

结果：

 

01

> db.guest.find({}, {"PATENT" : 1, "_id" : 1})

02

{ "_id" : ObjectId("52692c2a0b082a1bbb727d86"), "PATENT" : 3070801 }

03

{ "_id" : ObjectId("52692c2a0b082a1bbb727d87"), "PATENT" : 3070802 }

04

{ "_id" : ObjectId("52692c2a0b082a1bbb727d88"), "PATENT" : 3070803 }

05

{ "_id" : ObjectId("52692c2a0b082a1bbb727d89"), "PATENT" : "IL" }

06

{ "_id" : ObjectId("52692c2a0b082a1bbb727d8a"), "PATENT" : 3070804 }

07

{ "_id" : ObjectId("52692c2a0b082a1bbb727d8b"), "PATENT" : 3070805 }

08

{ "_id" : ObjectId("52692c2a0b082a1bbb727d8c"), "PATENT" : 3070806 }

09

{ "_id" : ObjectId("52692c2a0b082a1bbb727d8d"), "PATENT" : 3070807 }

10

{ "_id" : ObjectId("52692c2a0b082a1bbb727d8e"), "PATENT" : 3070808 }

11

{ "_id" : ObjectId("52692c2a0b082a1bbb727d8f"), "PATENT" : 3070809 }

12

> db.guest.count()

13

10

14

>

刚好10条，可见此命令导入是不会过滤异常数据。

2.以UTF-8有BOM格式再试一次。实验数据同上

 

01

> db.guest.find({}, {"PATENT" : 1, "_id" : 1})

02

{ "_id" : ObjectId("52692d730b082a1bbb727d90"), "PATENT" : 3070801 }

03

{ "_id" : ObjectId("52692d730b082a1bbb727d91"), "PATENT" : 3070802 }

04

{ "_id" : ObjectId("52692d730b082a1bbb727d92"), "PATENT" : 3070803 }

05

{ "_id" : ObjectId("52692d730b082a1bbb727d93"), "PATENT" : "IL" }

06

{ "_id" : ObjectId("52692d730b082a1bbb727d94"), "PATENT" : 3070804 }

07

{ "_id" : ObjectId("52692d730b082a1bbb727d95"), "PATENT" : 3070805 }

08

{ "_id" : ObjectId("52692d730b082a1bbb727d96"), "PATENT" : 3070806 }

09

{ "_id" : ObjectId("52692d730b082a1bbb727d97"), "PATENT" : 3070807 }

10

{ "_id" : ObjectId("52692d730b082a1bbb727d98"), "PATENT" : 3070808 }

11

{ "_id" : ObjectId("52692d730b082a1bbb727d99"), "PATENT" : 3070809 }

12

> db.guest.count()

13

10

结果同上面一样，key"PATENT "中并没有因BOM引起的空格

3.mongoimport命令解释

 

1

mongoimport -d TYK -c guest --type csv --file d:\text.csv --headerline

2

-d 数据库

3

-c 集合

4

--type 数据格式

5

--file 文件路径

6

--headerline 貌似指定这个后以第一行为key,另 -f 可以指定key “-f Name, age”

二、统计分析

 

1.根据性别统计

 

由于数据不规范，先查询一下有多少种方式来表示性别的

 

1

db.runCommand({"distinct" : "guestHouse", "key" : "Gender"})

01

{

02

  "values" : [

03

  "M",

04

  "F",

05

  "0",

06

  " ",

07

  "1",

08

  "",

09

  "19790522",

10

  "#0449",

11

  "#M",

12

  "

13   "N"

14   ],

15   "stats" : {

16   "n" : 20048891,

17   "nscanned" : 20048891,

18   "nscannedObjects" : 20048891,

19   "timems" : 377764,

20   "cursor" : "BasicCursor"

21   },

22   "ok" : 1

23 }

一共有11中方式表示性别的...那就以M、F做下统计吧

1 #总数据

2 db.guestHouse.count()

3 20048891

1 #男 M

2 db.guestHouse.count({"Gender":"M"})

3 12773070

4 64%

1 #女 F

2 db.guestHouse.count({"Gender":"F"})

3 6478745

4 32%

饼状图

 

总结：

 

1.带条件count时速度是非常慢的，猜测在count时可能先进行的查询操作，如果是查询加索引效果会好很多。对Gender加索引，效果明显提高了，但仍然是N秒级别的。显然在实时情况下还是不行的。另外随意加索引也会遇其它方面的问题。在用索引时能达到一个平衡点很重要的啊。

 

1 2013-10-24

2 查看count的js解释

3 > db.guestHouse.count

4 function ( x ){

5   return this.find( x ).count();

6 }

7 >

8 果然是先find,后count

2、根据身份证分析性别

 

从上面数据看，大约有4%的数据性别不详。

 

15位身份证号码：第7、8位为出生年份(两位数)，第9、10位为出生月份，第11、12位代表出生日期，第15位代表性别，奇数为男，偶数为女。 18位身份证号码：第7、8、9、10位为出生年份(四位数)，第11、第12位为出生月份，第13、14位代表出生日期，第17位代表性别，奇数为男，偶数为女。

要根据身份证来分析的话，明显不好直接处理分析了。那么就尝试一下编写MapReduce算一下吧，但是单机MapReduce速度会更慢。

先了解一下数据，看看有多少证件类型

 

01 > db.runCommand({"distinct" : "guestHouse", "key" : "CtfTp"})

02 {

03   "values" : [

04   "OTH",

05   "GID",

06   "ID",

07   "TBZ",

08   "VSA",

09   "TXZ",

10   "JID",

11   "JZ",

12   "HXZ",

13   "JLZ",

14   "#ID",

15   "hvz",

16   "待定",

17   "11",

18   "",

19   "SBZ",

20   "JGZ",

21   "HKB",

22   "TSZ",

23   "JZ1",

24   " ",

25   "Id",

26   "#GID",

27   "1"

28   ],

29   "stats" : {

30   "n" : 20048891,

31   "nscanned" : 20048891,

32   "nscannedObjects" : 20048891,

33   "timems" : 610036,

34   "cursor" : "BasicCursor"

35   },

36   "ok" : 1

37 }

38 >

数据依旧的乱，那就暂且以"ID"来统计一下吧

01 >map = function() {

02   if (this.CtfTp == "ID") {

03   if (this.CtfId.length == 18){

04   emit(parseInt(this.CtfId.charAt(16)) % 2, {count : 1}) //1为男,0为女

05   }else if (this.CtfId.length == 15) {

06   emit(parseInt(this.CtfId.charAt(14)) % 2, {count : 1}) //无法解析时为NaN

07   }

08   } else {

09   emit(-1, {count : 1})

10   }

11 }

12 

13 >reduce = function(key, emits) {

14   total = 0;

15   for (var i in emits) {

16   total += emits[i].count;

17   }

18   return {"count" : total};

19 }

20 

21 >mr = db.runCommand(

22  {

23  mapReduce: "guestHouse",

24  map: map,

25  reduce: reduce,

26  out: "TYK.guestHouse.output",

27  verbose: true

28  }

29  )

30 >{

31   "result" : "guestHouse.output",

32   "timeMillis" : 999097,

33   "timing" : {

34   "mapTime" : 777955,

35   "emitLoop" : 995248,

36   "reduceTime" : 111217,

37   "mode" : "mixed",

38   "total" : 999097

39   },

40   "counts" : {

41   "input" : 20048891,

42   "emit" : 19928098,

43   "reduce" : 594610,

44   "output" : 4

45   },

46   "ok" : 1

47 }

结果：

 

1 > db.guestHouse.output.find()

2 { "_id" : NaN, "value" : { "count" : 1360 } }

3 { "_id" : -1, "value" : { "count" : 1161164 } }

4 { "_id" : 0, "value" : { "count" : 6831007 } }

5 { "_id" : 1, "value" : { "count" : 11934567 } } 

6 >