hadoop项目实战–ETL–（二）实现自动向mysql中添加数据

**四

项目开发**

1
创建数据库
db_etl,
新建两张表
user
和
oder
。表结构如第一部分图所示。

2
编写
python
脚本，实现自动向
mysql
中插入数据。

新建python
项目，目录结构如下图

编写代码如下：

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
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
1# _*_ coding:UTF-8 _*_

2'''

3Created on 2016年12月1日

4

5@author: duking

6'''

7import MySQLdb

8import random,string

9import time

10import threading

11'''

12数据库连接

13'''

14def ConnMysql():

15    #连接数据库

16    conn = MySQLdb.connect(host = "192.168.0.154", user = 'root', passwd = '123456', db = 'db_etl', charset = 'utf8')

17    cursor = conn.cursor()

18    return conn,cursor

19

20'''

21插入user数据

22'''

23def AddUserInfo(username,passwd):

24    

25    conn,cursor = ConnMysql()

26    

27    sql = "insert into userinfo(username,passwd) values(%s,%s)"

28

29    param = (username,passwd)

30     

31    cursor.execute(sql,param)

32    

33    conn.commit()

34    cursor.close()

35    conn.close()

36    

37'''

38插入order数据

39'''

40def AddOderInfo(warename,price):

41    

42    conn,cursor = ConnMysql()

43    

44    sql = "insert into oderinfo(warename,price) values(%s,%s)"

45

46    param = (warename,price)

47     

48    cursor.execute(sql,param)

49    

50    conn.commit()

51    cursor.close()

52    conn.close()

53

54'''

55随机产生字符串

56'''

57def Random_Str(randomlength):

58    a = list(string.ascii_letters)

59    random.shuffle(a)

60    return ''.join(a[:randomlength])

61    

62

63#随机生成订单信息

64def MakeOderInfo(threadname):

65    while(True):

66        #随机10~100秒生成一条Oder信息

67        time.sleep(random.randint(10,100))   

68        AddOderInfo(Random_Str(random.randint(6,10)),float(round(random.uniform(10,100),2)))

69        print threadname + ':a new OderInfo is Maked    ' + time.ctime(time.time())

70

71#随机生成用户信息

72def MakeUserInfo(threadname):

73    while(True):

74        time.sleep(random.randint(20,100))

75        AddUserInfo(Random_Str(random.randint(6,10)),Random_Str(random.randint(6,10)))

76        print threadname + ':a new UserInfo is Maked    ' +time.ctime(time.time())

77    

78    

79#python 模块的入口：main函数

80if __name__ == '__main__':

81    

82    #多线程

83    thread_1 = threading.Thread(target=MakeOderInfo,args=('thread_1', ))

84    thread_2 = threading.Thread(target=MakeUserInfo,args=('thread_2', ))

85    

86    #启动线程

87    thread_1.start()

88    thread_2.start()

89        

90

91        

92

注意：python
调用
mysql
需要引入
MySQLdb
模块，改模块的安装请看另外的教程

最后，将写好的python
在
linux
中运行。

运行后查看数据库就可以看见数据在不断的增长了。