数据工程师常用的几个小工具（附python源代码）

 1 import MySQLdb
 2 import pandas as pd
 3 from sqlalchemy import create_engine
 4 
 5 class con_analyze:
 6     """数据分析平台连接"""
 7 
 8     def __init__(self, database='myanalyze'):
 9         self.database = database
10         self.conn = None
11 
12     def connect(self):
13         self.conn = MySQLdb.connect(host='***', user='root', passwd='***', db=self.database,
14                                     charset='utf8')
15 
16     def query(self, sql):
17         try:
18             self.connect()
19             data = pd.read_sql(sql, self.conn)
20         except (AttributeError, MySQLdb.OperationalError):
21             self.connect()
22             data = pd.read_sql(sql, self.conn)  # 读取数据出现错误,再次连接
23         return data
24 
25     def store(self, mydataframe, table_name, if_exists='replace'):
26         conn2 = "mysql+mysqldb://root:***@***:3306/%s" % self.database
27         local_engine = create_engine(conn2)
28         mydataframe.to_sql(table_name, local_engine, if_exists=if_exists, index=False, chunksize=10000)
29 
30     '''还可以加一个函数用来执行单条sql语句,不仅仅是读取数据，还可以update，create等'''

 1 from config import con_analyze
 2 
 3 
 4 class AnalyzeData:
 5     def __init__(self):
 6         # 此处初始化，可以带一个参数：database,默认为myanalyze
 7         self.conn = con_analyze()
 8         # self.conn2 = con_analyze("myanalyze_2")
 9 
10     def get_data(self, sql):
11         # 执行sql查询结果保存到df中
12         df = self.conn.query(sql=sql)
13 
14     def store_data(self, df):
15         # 将dataframe类型的数据df，存入名为dd_name的数据表中
16         self.conn.store(df, 'db_name')

1 import pymongo
 2 import pandas as pd
 3 
 4 class Conn_Mongo:
 5     """mongo 数据库连接"""
 6 
 7     def __init__(self):
 8         self.mongo_utoken = pymongo.MongoClient('mongodb://***:27000').utoken  # 用户表
 9 
10     def get_user_data_mongo(self,list_id):
11         """
12         通过连接 mongo查找 
13         """
14         user_data = pd.DataFrame(list(self.mongo_fotor.userinfo.find({'FToken': {'$in': list(list_id)}})))
15         return user_data

import pandas as pd
import json, requests

 1 class Conn_Flurry:
 2     """flurry api data"""
 3     api_token = "******.****.****" 
 4     headers = {'Authorization': 'Bearer {}'.format(api_token)}
 5     url = "https://api-metrics.flurry.com/public/v1/data/appEvent/day/app?metrics=activeDevices,newDevices,averageTimePerDevice&dateTime=2017-05-23/2017-05-24"
 6 
 7     def get_results(self, url=url):
 8         '''
 9         这里使用的url是一个示例，也可以使用get_url函数创建需要的url传入此函数作为参数
10         '''
11         data = requests.get(url, headers=self.headers)
12         cleaned = json.loads(data.text, 'utf-8')
13         cleaned = pd.DataFrame(cleaned['rows'])
14         return cleaned
15 
16     def get_url(self, table='appEvent', timegrain='day', dimensions='app/event', metrics='occurrences',
17                 dateTime='2017-09-23/2017-05-24', filters=""):
18         '''
19         若filters为空， 不影响结果
20         标准的url：endpoint + '/table/timeGrain/dimension1/dimension2;show=all/dimension3{...}?metrics=[comma-separated-metrics]&dateTime=[..]&filters=[...]&topN=[..]&sort=[..]&having=[..]&format=[..]&timeZone=[..]'
21         App Usage url： endpoint+ "/appUsage/day?metrics=sessions,activeDevices,newDevices&dateTime=2016-06-01/2016-08-01&filters=app|name-in[appname]"
22         app event url: endpoint + "/appEvent/day/app/appVersion/event?metrics=occurrences&dateTime=2016-07-01/2016-07-03&filters=app|name-in[foo],event|name-in[login,register]"
23         app event url2: endpoint + "/appEvent/day/app/country?metrics=activeDevices,newDevices&dateTime=2016-07-01/2016-07-03&filters=app|name-in[foo],event|name-in[login]&topN=5&sort=activeDevices|desc"
24         event parameter: endpoint+ "/eventParams/day/app;show=all/event/paramName/paramValue?metrics=count&dateTime=2016-11-07/2016-11-08&filters=app|name-in[foo],event|name-in[level_complete]"
25         注意，dimensions的变化，当要看某一事件的具体信息时：app;show=all/event/paramName/paramValue，加了个show=all
26         注意filters里面filters的格式，可以选择app名称和事件名称
27         注意timegrain和datetime的关系，常见的就是day和month，datetime的格式也要跟着变
28         '''
29         endpoint = 'https://api-metrics.flurry.com/public/v1/data'
30         url = "{}/{}/{}/{}?metrics={}&dateTime={}&filters={}".format(endpoint, table, timegrain, dimensions, metrics,
31                                                                      dateTime, filters)
32         return url

1 url = self.conn_flurry.get_url('appUsage', 'month', 'app','averageTimePerSession,activeDevices,newDevices,sessions', self.time_range)
2 user_mobile = self.conn_flurry.get_results(url)

self.time_range = '2017-09/2017-10'

'2017-09-23/2017-10-24'