数据工程师推荐你用的几个工具

作为数据工程师或者数据分析师,经常会跟各种数据打交道,其中,获取数据这一关是无法避免的,下面,我就将自己时常工作中用到的数据连接配置模型分享出来,供大家交流。

MySQL数据库

mysql数据库是目前用的最多的数据库之一,此处我做的是读和写的接口,而删除和更新操作,一般不是分析师做的,而是开发,所以我没有做这个。

import MySQLdb
import pandas as pd
from sqlalchemy import create_engine

class con_analyze:
 """数据分析平台连接"""

 def __init__(self, database='myanalyze'):
 self.database = database
 self.conn = None

 def connect(self):
 self.conn = MySQLdb.connect(host='***', user='root', passwd='***', db=self.database, charset='utf8')

 def query(self, sql):
 try:
 self.connect()
            data = pd.read_sql(sql, self.conn)
 except (AttributeError, MySQLdb.OperationalError):
 self.connect()
            data = pd.read_sql(sql, self.conn) # 读取数据出现错误,再次连接
 return data

 def store(self, mydataframe, table_name, if_exists='replace'):
        conn2 = "mysql+mysqldb://root:***@***:3306/%s" % self.database
        local_engine = create_engine(conn2)
        mydataframe.to_sql(table_name, local_engine, if_exists=if_exists, index=False, chunksize=10000)

 '''还可以加一个函数用来执行单条sql语句,不仅仅是读取数据,还可以update,create等'''

作为一个链接类来使用,初始化的时候给出的conn是None,只有在执行查询函数的时候才创建链接,(链接中,我隐去了自己的host信息,你需要将自己的host填进去)

查询的时候使用了try语句,如果链接不成功或者查询不成功,就会出错,如果是链接不成功,那就在异常中再次连接。关于重复执行一段代码,有一个库大家可以关注一下:tenacity 这个库能让你实现更优雅(pythonic)的代码重复 此处读取数据是使用pandas库中的read_sql函数,此函数可以直接将查询结果转化成一个dataframe,方便了后面的分析工作

存储功能也是使用dataframe的函数tosql,此函数是将一个df直接转化成sql数据存入数据库,如果tablename存在,可以选择替换(replace)、增加(append)等,如果df很大很长,就需要设置一下chunksize参数

chunksize的设定,程序会自动将你的长达几十万行的df迭代存储,每次只存储10000行(这个数字是我设定的,你也可以改)。

看到这里,你可能会有疑问,为什么读和写的conn不一样,一个是用 MySQLdb.connect创建,而另一个是用create_engine创建。我想说的是,后面这个conn2其实可以作为读的连接参数,但是使用 MySQLdb.connect创建的连接却不一定能用来写,因为我在实践中多次运行发生了错误,所以我就改了。

其实,其他的数据库可以类似这种做法,给自己的项目配置一个连接类,使用的时候应该是这样的:

首先,你需要把代码放在一个单独的配置文件,比如config.py中

然后在你需要使用的地方,导入此配置文件

from config import con_analyze


class AnalyzeData:
 def __init__(self):
 # 此处初始化,可以带一个参数:database,默认为myanalyze
 self.conn = con_analyze()
 # self.conn2 = con_analyze("myanalyze_2")

 def get_data(self, sql):
 # 执行sql查询结果保存到df中
        df = self.conn.query(sql=sql)

 def store_data(self, df):
 # 将dataframe类型的数据df,存入名为dd_name的数据表中
 self.conn.store(df, 'db_name')

MongoDB

mongodb是一个非结构化数据库,里面存储的数据类似于json,是键值对的形式,如果你遇到了需要查询mongodb中的数据,下面我就简单介绍一下。

同样,也是要建立一个类,这是为了规范。

import pymongo
import pandas as pd

class Conn_Mongo:
 """mongo 数据库连接"""

 def __init__(self):
 self.mongo_utoken = pymongo.MongoClient('mongodb://***:27000').utoken  # 用户表

 def get_user_data_mongo(self,list_id):
 """
        通过连接 mongo查找 
        """
        user_data = pd.DataFrame(list(self.mongo_fotor.userinfo.find({'FToken': {'$in': list(list_id)}})))
 return user_data

这个毕竟简单,就是一个查询操作,我是先传入一串id,根据id找到对应的信息。一般来说,mongodb的库容量都比较大,所以我是有针对的查询相关信息。

这里用到了pymongo库,通过它创建一个到相应地址(我用*隐掉了)的连接,后面的.utoken是对应的库名称,其实你也可以把它作为参数,在初始化的时候传进去。

后面查询的时候使用了find函数,其前面的userinfo是表的名称,find的参数也是键值对的形式,这里我指定了键的名称"FToken",其值{'$in': list(list_id)}代表的意思是:在什么什么中。

将id 做成了一个list(为了大家理解,取名为list_id),相关语法大家可以查阅一下。

Flurry

如果你的工作涉及到了app的数据,那经常会使用Flurry获取数据。

Flurry是一个移动统计平台,虽然是国外的,但国内依然可以用(不像谷歌分析被禁了),ios和Android应用的运营数据都可以在上面统计查询。

如果你还没有,又想了解的,可以戳这里:Flurry

对,网页浏览的话,界面就是这样的。

常用的功能是用户数据

以及功能点击事件

不过,这不是我要说的重点,上面只是让你看一下Flurry长什么样,现在我要写python接口,将这些数据取出。

Flurry的api地址,请戳这里:Flurry API

这是创建分析报告的api,有别于开发的api

首先,我们需要去申请一个app token,用于获取连接权限,申请方法请参考:app access token

它是大一串字母

只要获取到了这个token,我们就可以创建一个url,用于获取Flurry里面的数据了,具体看如下的代码:

import pandas as pd
import json, requests
class Conn_Flurry:
 """flurry api data"""
    api_token = "******.****.****" 
    headers = {'Authorization': 'Bearer {}'.format(api_token)}
    url = "https://api-metrics.flurry.com/public/v1/data/appEvent/day/app?metrics=activeDevices,newDevices,averageTimePerDevice&dateTime=2017-05-23/2017-05-24"

 def get_results(self, url=url):
 '''
        这里使用的url是一个示例,也可以使用get_url函数创建需要的url传入此函数作为参数
        '''
        data = requests.get(url, headers=self.headers)
        cleaned = json.loads(data.text, 'utf-8')
        cleaned = pd.DataFrame(cleaned['rows'])
 return cleaned

 def get_url(self, table='appEvent', timegrain='day', dimensions='app/event', metrics='occurrences',
                dateTime='2017-09-23/2017-05-24', filters=""):
 '''
        若filters为空, 不影响结果
        标准的url:endpoint + '/table/timeGrain/dimension1/dimension2;show=all/dimension3{...}?metrics=[comma-separated-metrics]&dateTime=[..]&filters=[...]&topN=[..]&sort=[..]&having=[..]&format=[..]&timeZone=[..]'
        App Usage url: endpoint+ "/appUsage/day?metrics=sessions,activeDevices,newDevices&dateTime=2016-06-01/2016-08-01&filters=app|name-in[appname]"
        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]"
        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"
        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]"
        注意,dimensions的变化,当要看某一事件的具体信息时:app;show=all/event/paramName/paramValue,加了个show=all
        注意filters里面filters的格式,可以选择app名称和事件名称
        注意timegrain和datetime的关系,常见的就是day和month,datetime的格式也要跟着变
        '''
        endpoint = 'https://api-metrics.flurry.com/public/v1/data'
        url = "{}/{}/{}/{}?metrics={}&dateTime={}&filters={}".format(endpoint, table, timegrain, dimensions, metrics,
                                                                     dateTime, filters)
 return url

代码稍微有点长,中间许多注释行,但总的来说就是两个步骤:

1. 构建url

2. 获取url对应的结果

但是细细说来,这里面涉及到的东西比较多,比如,为什么url的格式是这样的,还有headers为什么是那样构造的,还有结果的形式等等

我想说的是,这些在官网api上已有很详细的说明,我就不搬砖了,不过,如果你有任何疑问,欢迎在评论区留言,我知道的一定尽心解答。

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

上面就是一个简单的应用,其中time_range应该是这样的格式

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

对于这个时间范围,Flurry默认是左闭右开的,即不包含10月

同理,如果是这样

那就代表从9月23号起,但是不包含10月24号的结果,这一点尤其要注意。如果你是拿某一段时间内的数据,就很容易忽略这点,导致少拿数据

如果是按天拿还好,有date这个维度,会提醒你到底拿到了哪些天的数据。

原文发布于微信公众号 - IT派(transfer_3255716726)

原文发表时间:2018-05-16

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏Spark生态圈

spark任务之Task失败监控

在spark程序中,task有失败重试机制(根据 spark.task.maxFailures 配置,默认是4次),当task执行失败时,并不会直接导致整个应用...

4093
来自专栏MixLab科技+设计实验室

设计师编程指南之Sketch插件开发 1

发现网上关于sketch插件开发的指南太少了,而且都不一定可以成功运行,于是我就写了这个系列的文章: 1 我们需要了解的语法特点 sketch 是基于 Coc...

5288
来自专栏黑白安全

PHP代码审计入门之路

虽然市面上的代码审计的文章已经一大把了,但是还是决定重复造轮子,打算作为一个系列来写的,近年越来越多的安全研究人员投入到php应用的漏洞挖掘,相对应的代码安全问...

932
来自专栏游戏杂谈

关于seajs

虽然已经有很长时间没写JavaScript,但很多时候看到一些应用还是会带着好奇心去研究一下。之前是看腾讯的朋友网,它的webchat做的很不错(虽然ff下有b...

2093
来自专栏Golang语言社区

论获取缓存值的正确姿势

论获取缓存值的正确姿势 cache 时至今日,大家对缓存想必不在陌生。我们身边各种系统中或多或少的都存在缓存,自从有个缓存,我们可以减少很多计算压力,提高应用程...

3728
来自专栏码农阿宇

海量数据即时查询引擎ElasticSearch入门 附.Net Core例子

Elasticsearch是基于Lucene的搜索引擎。它提供了一个分布式,支持多租户的全文搜索引擎,它具有HTTP Web界面和无模式JSON文档。 Elas...

1972
来自专栏Java 源码分析

Netty 入门

1. 粘包问题 一 .长连接与短连接: 1.长连接:Client方与Server方先建立通讯连接,连接建立后不断开, 然后再进行报文发送和接收。长连接在 net...

2955
来自专栏Java帮帮-微信公众号-技术文章全总结

大文件拆分方案的Java实践【面试+工作】

大文件拆分问题涉及到io处理、并发编程、生产者/消费者模式的理解,是一个很好的综合应用场景,为此,花点时间做一些实践,对相关的知识做一次梳理和集成,总结一些共性...

2274
来自专栏码农阿宇

ElasticSearch入门 附.Net Core例子

Elasticsearch是基于Lucene的搜索引擎。它提供了一个分布式,支持多租户的全文搜索引擎,它具有HTTP Web界面和无模式JSON文档。 Elas...

2801
来自专栏CDA数据分析师

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

? 作者 小溏 原文链接:http://www.cnblogs.com/lihuafengzi/p/8243904.html 作为数据工程师或者数据分析师...

2407

扫码关注云+社区