forked from mining/mining
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathcube.py
More file actions
executable file
·182 lines (138 loc) · 5.32 KB
/
Copy pathcube.py
File metadata and controls
executable file
·182 lines (138 loc) · 5.32 KB
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
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import json
import riak
import gc
import traceback
from datetime import datetime
from pandas import DataFrame
from sqlalchemy import create_engine
from sqlalchemy.sql import text
from sqlalchemy.orm import sessionmaker
from mining.utils import conf, log_it
from mining.utils._pandas import fix_render, CubeJoin
from mining.multithread import ThreadPool
from bottle.ext.mongo import MongoPlugin
def run(cube_slug=None):
mongo = MongoPlugin(
uri=conf("mongodb")["uri"],
db=conf("mongodb")["db"],
json_mongo=True).get_mongo()
pool = ThreadPool(20)
for cube in mongo['cube'].find():
slug = cube['slug']
if cube_slug and cube_slug != slug:
continue
pool.add_task(process, cube)
pool.wait_completion()
return True
class CubeProcess(object):
def __init__(self, _cube):
log_it("START: {}".format(_cube['slug']), "bin-mining")
self.mongo = MongoPlugin(
uri=conf("mongodb")["uri"],
db=conf("mongodb")["db"],
json_mongo=True).get_mongo()
MyClient = riak.RiakClient(
protocol=conf("riak")["protocol"],
http_port=conf("riak")["http_port"],
host=conf("riak")["host"])
self.MyBucket = MyClient.bucket(conf("riak")["bucket"])
self.MyBucket.enable_search()
del _cube['_id']
self.cube = _cube
self.slug = self.cube['slug']
def load(self):
self.cube['run'] = 'run'
self.mongo['cube'].update({'slug': self.slug}, self.cube)
self.cube['start_process'] = datetime.now()
_sql = self.cube['sql']
if _sql[-1] == ';':
_sql = _sql[:-1]
self.sql = u"""SELECT * FROM ({}) AS CUBE;""".format(_sql)
self.connection = self.mongo['connection'].find_one({
'slug': self.cube['connection']})['connection']
log_it("CONNECT IN RELATION DATA BASE: {}".format(self.slug),
"bin-mining")
e = create_engine(self.connection,
**conf('openmining')['sql_conn_params'])
Session = sessionmaker(bind=e)
session = Session()
resoverall = session.execute(text(self.sql))
self.data = resoverall.fetchall()
self.keys = resoverall.keys()
def environment(self, t):
if t not in ['relational']:
self.sql = t
def _data(self, data):
self.data = data
def _keys(self, keys):
self.keys = keys
def frame(self):
log_it("LOAD DATA ON DATAWAREHOUSE: {}".format(self.slug),
"bin-mining")
self.df = DataFrame(self.data)
if self.df.empty:
log_it('[warning]Empty cube: {}!!'.format(self.cube),
"bin-mining")
return
self.df.columns = self.keys
self.df.head()
self.pdict = map(fix_render, self.df.to_dict(outtype='records'))
def clean(self):
log_it("CLEAN DATA (JSON) ON RIAK: {}".format(self.slug),
"bin-mining")
self.MyBucket.new(self.slug, data='').store()
self.MyBucket.new(u'{}-columns'.format(self.slug), data='').store()
self.MyBucket.new(u'{}-connect'.format(self.slug), data='').store()
self.MyBucket.new(u'{}-sql'.format(self.slug), data='').store()
def save(self):
self.clean()
log_it("SAVE DATA (JSON) ON RIAK: {}".format(self.slug),
"bin-mining")
self.MyBucket.new(self.slug, data=self.pdict,
content_type="application/json").store()
log_it("SAVE COLUMNS ON RIAK: {}".format(self.slug),
"bin-mining")
self.MyBucket.new(u'{}-columns'.format(self.slug), data=json.dumps(
self.keys)).store()
log_it("SAVE CONNECT ON RIAK: {}".format(self.slug),
"bin-mining")
self.MyBucket.new(u'{}-connect'.format(self.slug),
data=self.connection).store()
log_it("SAVE SQL ON RIAK: {}".format(self.slug),
"bin-mining")
self.MyBucket.new(u'{}-sql'.format(self.slug), data=self.sql).store()
self.cube['status'] = True
self.cube['lastupdate'] = datetime.now()
self.cube['run'] = True
self.mongo['cube'].update({'slug': self.cube['slug']}, self.cube)
log_it("CLEAN MEMORY: {}".format(self.slug), "bin-mining")
gc.collect()
def process(_cube):
try:
log_it("START: {}".format(_cube['slug']), "bin-mining")
mongo = MongoPlugin(
uri=conf("mongodb")["uri"],
db=conf("mongodb")["db"],
json_mongo=True).get_mongo()
c = CubeProcess(_cube)
if _cube.get('type') == 'relational':
c.load()
c.frame()
c.save()
elif _cube.get('type') == 'cube_join':
c.environment(_cube.get('type'))
cube_join = CubeJoin(_cube)
c._data(cube_join.none())
c._keys(cube_join.none().columns.values)
c.frame()
c.save()
except Exception, e:
log_it(e, "bin-mining")
log_it(traceback.format_exc(), "bin-mining")
_cube['run'] = False
mongo['cube'].update({'slug': _cube['slug']}, _cube)
log_it("END: {}".format(_cube['slug']), "bin-mining")
if __name__ == "__main__":
run()