#查询用户余额代码案例

import sys

import MySQLdb

import pandas as pd

optmap = {

'dbuser' : 'aduser',

'dbpass' : '123654',

'dbhost' : '192.168.10.14',

'dbport' : 3306,

'dbname' : 'HBAODB'

}

def sql_select(reqsql):

try:

db_conn = MySQLdb.connect(user=optmap['dbuser'], passwd=optmap['dbpass'], host=optmap['dbhost'], port=optmap['dbport'], db=optmap['dbname'])

db_cursor=db_conn.cursor()

db_conn.query("use %s"%optmap['dbname'])

count = db_cursor.execute(reqsql)

ret = db_cursor.fetchall()

db_cursor.close()

db_conn.close

return ret

except MySQLdb.Error,e:

print "Mysql ERROR %d:%s" %(e.args[0], e.args[1])

return ''

def getusercoin(userid):

i = int(userid) % 10

reqsql = "select ID,COINREFER from CHARCOIN%u where ID=%u" % (int(i), int(userid))

#print reqsql

ret = sql_select(reqsql) #调用前面的函数

#print ret

return ret[0]

def getall(userlist):

userdata = pd.DataFrame(columns=('userid', 'coin'))

index = 0

for userid in userlist:

coins = getusercoin(userid) #调用前面的函数

#print coins[0],coins[1]/100.0

if coins[0] is not None:

userdata.loc[index] = (str(coins[0]), coins[1]/100.0)

else:

userdata.loc[index] = (str(userid), 0)

index += 1

#print userdata.tail(10)

df = spark.createDataFrame(userdata)

#df.createOrReplaceTempView('userdata')

df.show(50)

#用户消费查询代码案例

import sys

import MySQLdb

import pandas as pd

import datetime

import time

optmap = {

'dbuser' : 'aduser',

'dbpass' : '123654',

'dbhost' : '192.168.10.12',

'dbport' : 3306,

'dbname' : 'JIESUANDB'

}

def sql_select(reqsql):

try:

db_conn = MySQLdb.connect(user=optmap['dbuser'], passwd=optmap['dbpass'], host=optmap['dbhost'], port=optmap['dbport'], db=optmap['dbname'])

db_cursor=db_conn.cursor()

db_conn.query("use %s"%optmap['dbname'])

count = db_cursor.execute(reqsql)

ret = db_cursor.fetchall()

db_cursor.close()

db_conn.close

return ret

except MySQLdb.Error,e:

print "Mysql ERROR %d:%s" %(e.args[0], e.args[1])

return ''

#用户人民币消费

def getuserconsume(userid, startday): #定义带参函数

strdate = startday.strftime("%y%m%d")

# 送礼物 + 守护 + 点歌 + 表情贴

reqsql = "select CONSUMERID,SUM(DUBIOPNUM) from `DUBIJIESUANTONGJI_%s` where CONSUMERID=%u AND DBIOPTYPE=2 AND (OPTYPE=1 OR OPTYPE=4 OR OPTYPE=17 OR OPTYPE=25)" % (strdate, int(userid))

print reqsql

ret = sql_select(reqsql) #调用前面的函数

#print ret

if ret[0][0] is not None:

return float(ret[0][1])/100.0

else:

return 0

#用户充值

def getusercharge(userid, startday):

strdate = startday.strftime("%y%m%d")

reqsql = "select CONSUMERID,SUM(DUBIOPNUM) from `DUBIJIESUANTONGJI_%s` where CONSUMERID=%u AND DUBIOPTYPE=1 AND (OPTYPE=1016 OR OPTYPE=1020 OR OPTYPE=1021)" % (strdate, int(userid))

print reqsql

ret = sql_select(reqsql)#调用前面的函数

print ret

if ret[0][0] is not None:

return float(ret[0][1])/100.0

else:

return 0

#用户当天结余人民币

def getusercurcoin(userid, startday):

strdate = startday.strftime("%y%m%d")

reqsql = "select CONSUMERID,CURRENTNUM from `DUBIJIESUANTONGJI_%s` where CONSUMERID=%u ORDER BY OPTIME DESC LIMIT 1" % (strdate, int(userid))

print reqsql

ret = sql_select(reqsql)

print ret

if ret:

return float(ret[0][1])/100.0

else:

return 0

def getconsume():

startdate = datetime.date(2017, 1, 1)

enddate = datetime.date(2017, 2, 2)

userid = 3101011990

userdata= pd.DataFrame(columns=('date','userid','charge', 'consume', 'dayleftcoin'))

index = 0

# 计算日差

td = enddate - startdate

datelen = td.days + 1

#print datelen

delta = datetime.timedelta(days=1)

allcoins = 0

for i in range(0,datelen):

startday = startdate + delta * i

consume_coin = getuserconsume(userid, startday)#调用前面的函数

charge = getusercharge(userid, startday)#调用前面的函数

dayleftcoin = getusercurcoin(userid, startday)#调用前面的函数

userdata.loc[index] = (startday.strftime("%Y-%m-%d"),str(userid), charge, consume_coin, dayleftcoin)

index += 1

#userdata.loc[index] = ('total',str(userid), allcoins, 0)

print userdata.tail(100)

return

getconsume()

#查询用户机器ID 代码案例

import sys

import MySQLdb

import pandas as pd

import datetime

optmap = {

'dbuser' : 'aduser',

'dbpass' : '123654',

'dbhost' : '192.168.10.15',

'dbport' : 3306,

'dbname' : 'JIQIDB'

}

def sql_select(reqsql):

try:

db_conn = MySQLdb.connect(user=optmap['dbuser'], passwd=optmap['dbpass'], host=optmap['dbhost'], port=optmap['dbport'], db=optmap['dbname'])

db_cursor=db_conn.cursor()

db_conn.query("use %s"%optmap['dbname'])

count = db_cursor.execute(reqsql)

ret = db_cursor.fetchall()

db_cursor.close()

db_conn.close

return ret

except MySQLdb.Error,e:

print "Mysql ERROR %d:%s" %(e.args[0], e.args[1])

return ''

def getusermid(userid, months):

i = int(userid) % 50

reqsql = "select USERID,MACHINEID from LOGINHISTORY%s%u where USERID=%u group by MACHINEID" % (months,int(i), int(userid))

print reqsql

ret = sql_select(reqsql)

#print ret

#print ret[0]

return ret

def getall(userlist):

today = datetime.date.today()

months = today.strftime("%Y%m")

userdata= pd.DataFrame(columns=('USERID', 'MACHINEID'))

index = 0

for userid in userlist:

coins = getusermid(userid, months)

for i in range(len(coins)):

#print coins[i]

userdata.loc[index] = (str(coins[i][0]), str(coins[i][1]))

index += 1

#print coins[0],coins[1]/100.0

#userdata.loc[index] = (str(coins[0]), coins[1]/100.0)

#index += 1

#print userdata.tail(10)

df = spark.createDataFrame(userdata)

#df.createOrReplaceTempView('userdata')

df.show(1000)

#人民币统计代码案例

from pyspark.sql import Row

from pyspark.sql.types import *

from pyspark.sql.functions import udf

import MySQLdb

import mysql_op

import datetime

import time

from mysql_op import MySQL

import pandas as pd

import numpy as np

from fastparquet import ParquetFile

from fastparquet import write

def fromDayToDay(startdate, datelen, func):

delta = datetime.timedelta(days=1)

for i in range(0,datelen):

startday = startdate + delta * i

endday = startdate + delta * (i + 1)

func(startday, endday)

return

def fromDayToEndDay(startdate, datelen, endday, func):

delta = datetime.timedelta(days=1)

for i in range(0,datelen):

startday = startdate + delta * i

#endday = startdate + delta * (i + 1)

func(startday, endday)

return

# 获取人民币数据

def saveDayPackageData(startday, endday):

#数据库连接参数

dbconfig = {'host':'192.168.10.12',

'port': 3306,

'user':'user',

'passwd':'123654',

'db':'JIESUANDB',

'charset':'utf8'}

#连接数据库，创建这个类的实例

mysql_cn= MySQLdb.connect(host=dbconfig['host'], port=dbconfig['port'],user=dbconfig['user'], passwd=dbconfig['passwd'], db=dbconfig['db'])

strday = startday.strftime("%Y%m%d")

tsstart=time.mktime(startday.timetuple())

tsend=time.mktime(endday.timetuple())

strdate = startday.strftime("%y%m%d")

sql = "SELECT OPTIME,CONSUMERID,PRESERVENUM,CURRENTNUM,DUBIOPTYPE,DUBIOPNUM,OPTYPE,OPDETAIL,OPNUM FROM `DUBIJIESUANTONGJI_%s`" % (strdate)

print sql

pddf = pd.read_sql(sql, con=mysql_cn)

mysql_cn.close()

print pddf.head(5)

dflen = len(pddf.index)

if dflen > 0:

print pddf.describe()

write("/home/haoren/logstatis/billdata"+strday+".parq", pddf)

return

def savePackageData():

startday = datetime.date(2016, 12, 28)

endday = datetime.date(2016, 12, 28)

td = endday - startday

datelen = td.days + 1

# 获取包裹数据

fromDayToDay(startday, datelen, saveDayPackageData)

# 获取WF册数据

def saveDayWifiPhoneRegData(startday, endday):

#数据库连接参数

dbconfig = {'host':'192.168.10.15',

'port': 3306,

'user':'user',

'passwd':'123654',

'db':'AADB',

'charset':'utf8'}

#连接数据库，创建这个类的实例

mysql_cn= MySQLdb.connect(host=dbconfig['host'], port=dbconfig['port'],user=dbconfig['user'], passwd=dbconfig['passwd'], db=dbconfig['db'])

strday = startday.strftime("%Y%m%d")

tsstart=time.mktime(startday.timetuple())

tsend=time.mktime(endday.timetuple())

strdate = startday.strftime("%y%m%d")

sql = "select USERID from NEW_WEB_USER where TIME< %d AND TYPE=17" % (tsend)

print sql

pddf = pd.read_sql(sql, con=mysql_cn)

mysql_cn.close()

print pddf.head(5)

dflen = len(pddf.index)

if dflen > 0:

print pddf.describe()

write("/home/haoren/logstatis/wifiphonereg"+strday+".parq", pddf)

return

def saveWifiPhoneReg():

startday = datetime.date(2016, 12, 1)

endday = datetime.date(2016, 12, 1)

td = endday - startday

datelen = td.days + 1

# 获取包裹数据

fromDayToDay(startday, datelen, saveDayWifiPhoneRegData)

OPTypeName = {

0:"会员",

1:"道具",

}

OpDetailName19 = {

1:"购物保存收益",

2:"下注和返注",

3:"发红包",

4:"抢红包",

}

OpDetailName22 = {

1:"活动1收益到总账号",

2:"活动2收益到总账号",

3:"活动3收益到总账号",

}

OpDetailName23 = {

0:"购买会员",

1:"购买道具",

2:"扫雷",

}

def getOpTypeName(func):

name = OPTypeName.get(func)

if name == None:

return ""

else:

return name.decode('utf8')

def getOpDetailName(func, detail):

if func == 19:

if detail > 10000 and detail < 30000:

return "包裹回滚".decode('utf8')

elif detail > 50000 and detail < 60000:

return "红包接龙".decode('utf8')

else:

name = OpDetailName19.get(detail)

if name == None:

return ""

else:

return name.decode('utf8')

elif func == 22:

name = OpDetailName22.get(detail)

if name == None:

return ""

else:

return name.decode('utf8')

elif func == 23:

name = OpDetailName23.get(detail)

if name == None:

return ""

else:

return name.decode('utf8')

else:

return ""

def getDayPackageData(startday, endday):

strday = startday.strftime("%Y%m%d")

print strday + '人民币数据'

df = spark.read.load("/home/haoren/logstatis/billdata"+strday+".parq")

df.show(10)

#df.createOrReplaceTempView('billdata')

#df.registerTempTable("billdata")

#sqlret = sqlc.sql("SELECT count(*) from billdata")

#sqlret.show(1)

df2 = df.withColumn('OPTYPENAME', udf(getOpTypeName)(df.OPTYPE))

df2.show(10)

df = df2.withColumn('DETAILNAME', udf(getOpDetailName)(df2.OPTYPE, df2.OPDETAIL))

df.show(10)

df.createOrReplaceTempView('billdata')

return

def getPackageData():

startday = datetime.date(2016, 12, 28)

endday = datetime.date(2016, 12, 28)

td = endday - startday

datelen = td.days + 1

# 获取包裹数据

fromDayToDay(startday, datelen, getDayPackageData)#调用前面的函数

print 'getPackageData finish'

# 获取充值数据

def getChargeInfo(startday, endday):

#数据库连接参数

dbconfig = {'host':'192.168.10.14',

'port': 3306,

'user':'user',

'passwd':'123654',

'db':'BAOIMDB',

'charset':'utf8'}

#连接数据库，创建这个类的实例

mysql_cn= MySQLdb.connect(host=dbconfig['host'], port=dbconfig['port'],user=dbconfig['user'], passwd=dbconfig['passwd'], db=dbconfig['db'])

strday = startday.strftime("%Y%m%d")

tsstart=time.mktime(startday.timetuple())

tsend=time.mktime(endday.timetuple())

regdata = pd.DataFrame()

for i in range(0, 20):

sql = "SELECT * FROM `USERCONSUMPTIONRECORD%d` where TIME > %d AND TIME < %d" % (i, tsstart, tsend)

print sql

#pddf = pd.DataFrame()

pddf = pd.read_sql(sql, con=mysql_cn)

#print pddf.head(5)

if len(pddf.index) > 0:

regdata = regdata.append(pddf,ignore_index=True)

print regdata.tail(5)

if len(regdata.index) > 0:

print regdata.describe()

write("/home/haoren/logstatis/register"+strday+".parq", regdata)

mysql_cn.close()

return

def pudf(x):

return getOpTypeName(x.OPTYPE)

def getMergeData(strday):

dfbill = ParquetFile("/home/haoren/logstatis/billdata"+strday+".parq").to_pandas()

dfwifireg = ParquetFile("/home/haoren/logstatis/wifiphonereg"+strday+".parq").to_pandas()

tempdf = pd.merge(dfbill, dfwifireg, left_on='CONSUMERID', right_on='USERID')

#write("/home/haoren/logstatis/analyze"+strday+".parq", tempdf)

#print tempdf.head(10)

tempdf['OPTYPENAME'] = tempdf.apply(lambda x:getOpTypeName(x.OPTYPE), axis=1)

#print tempdf.head(10)

tempdf['DETAILNAME'] = tempdf.apply(lambda x:getOpDetailName(x.OPTYPE,x.OPDETAIL), axis=1)

df = spark.createDataFrame(tempdf)

df.show(10)

return df

def analyzeDayBillData(startday, endday):

strday = startday.strftime("%Y%m%d")

print strday + '人民币数据'

df = spark.read.load("/home/haoren/logstatis/billdata"+strday+".parq")

dfwifireg = spark.read.load("/home/haoren/logstatis/wifiphonereg"+strday+".parq")

df3 = df.join(dfwifireg, df.CONSUMERID == dfwifireg.USERID)

df3.show(10)

df3.write.parquet("/home/haoren/logstatis/analyze"+strday+".parq")

#df2 = df3.withColumn('OPTYPENAME', udf(getOpTypeName)(df3.OPTYPE))

#df2.show(10)

#df = df2.withColumn('DETAILNAME', udf(getOpDetailName)(df2.OPTYPE, df2.OPDETAIL))

#df.show(10)

#df.createOrReplaceTempView('analyzebilldata')

return

def analyzeDayBillData2(startday, endday):

strday = startday.strftime("%Y%m%d")

print strday + '人民币数据'

#df = spark.read.load("/home/haoren/logstatis/analyze"+strday+".parq")

df = getMergeData(strday)

return

df2 = df.withColumn('OPTYPENAME', udf(getOpTypeName)(df.OPTYPE))

df2.show(10)

df = df2.withColumn('DETAILNAME', udf(getOpDetailName)(df2.OPTYPE, df2.OPDETAIL))

df.show(10)

df.createOrReplaceTempView('analyzebilldata')

return

def analyzeBillData():

startday = datetime.date(2016, 12, 28)

endday = datetime.date(2016, 12, 28)

td = endday - startday

datelen = td.days + 1

# 获取包裹数据

fromDayToDay(startday, datelen, analyzeDayBillData2)#调用前面的函数

print 'analyzeBillData finish'

savePackageData()

getPackageData()

#saveWifiPhoneReg()

#analyzeBillData()

python大数据分析实例-python大数据分析代码案例相关推荐

1. python装饰器实例-python装饰器实例大详解

   原标题:python装饰器实例大详解 一.作用域 在python中,作用域分为两种:全局作用域和局部作用域. 全局作用域是定义在文件级别的变量,函数名.而局部作用域,则是定义函数内部. 关于作用域,我 ...

2. python数据分析实例-python数据分析实战

   Python数据分析实战(高清版)PDF 百度网盘 链接:https://pan.baidu.com/s/1nlHM1IW8MYg3z79TUwIsWg 提取码:ux8t 复制这段内容后打开百度网盘手 ...

3. 247个Python经典有趣实例，185页代码齐全可复制，PDF版拿走即用

   如今,Python的发展如日中天,在市场上占据了很大一块份额,越来越多的人开始学习Python,渴望通过Python达到自己的人生目标. 而学习Python的朋友都知道,学习Python的最佳方法就是 ...

4. python爬虫图片实例-Python爬虫爬取煎蛋网图片代码实例

   这篇文章主要介绍了Python爬虫爬取煎蛋网图片代码实例,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下 今天,试着爬取了煎蛋网的图片. 用到的包: ...

5. python爬虫入门实例-Python爬虫天气预报实例详解（小白入门）

   本文研究的主要是Python爬虫天气预报的相关内容,具体介绍如下. 要求是把你所在城市过去一年的历史数据爬出来. 分析网站 我们可以看到,我们需要的天气数据都是放在图表上的,在切换月份的时候,发现只有 ...

6. python多线程爬虫实例-Python多线程爬虫简单示例

   python是支持多线程的,主要是通过thread和threading这两个模块来实现的.thread模块是比较底层的模块,threading模块是对thread做了一些包装的,可以更加方便的使用. ...

7. python简单程序实例-python简单项目实例

   语言多元化是PayPal编程文化中一个重要的组成部分.在C++和Java长期流行的同时,更多的团队选择了Jva和Scala.同时,Braintree的收购也引入了一个久经世故的Ruby社区.Pytho ...

8. python基础程序实例-Python基础的初始

   1.初始Python 1. pythonde 发展历史 自从20世纪90年代初Python语言诞生至今,它已被逐渐广泛应用于系统管理任务的处理和Web编程. Python的创始人为荷兰人吉多·范罗苏姆 ...

9. python简单程序实例-python简单实例训练(21~30)

   注意:我用的python2.7,大家如果用Python3.0以上的版本,请记得在print()函数哦!如果因为版本问题评论的,不做回复哦!! 21.题目:将一个正整数分解质因数.例如:输入90,打印出 ...

最新文章

1. The substring() Method in JDK 6 and JDK 7
2. android 点击图片事件,android图文混排点击事件
3. 中国存储器行业应用趋势与投资机遇研究报告2022版
4. PHP probuf详细步骤_go+protobuf+php简单示例
5. [转] MemCached 的 stats 命令
6. JVM系列五:JVM监测工具[整理中]
7. mp3播放程序c语言,Go语言音乐播放器
8. qwidget show 是否有信号_PyQt5信号与槽机制入门（一）
9. 基于卷积神经网络的草莓病害检测（新数据+基础算法？=SCI）
10. 传媒大学媒体中心资源批量获取工具的制作
11. kali破解压缩包密码
12. Redis cluster Hget hash field value异常分析案例--一个空格带来的惨案
13. 苹果系统打不开html,苹果6s的蜂窝移动数据打开没反应怎么办
14. xbox360 FSD 安装游戏教程(Freestyle3)
15. 理财产品收益计算公式揭密，破解6种获利模式
16. Ubuntu16.04 笔记本合上盖子时不进入休眠
17. 如何做到四位验证码更简洁的生成？
18. HTTP状态码全部说明
19. 用matlab实现用Bp神经网络对iris数据集进行分类（以及影响分类性能的参数条件）
20. 鸿蒙要出来了，安卓开发工程师有影响吗？

热门文章

1. 修改文件vim 插件：perl-support的修改和使用
2. Mark Links@2012/8/25
3. 剑指offer(19): 顺时针打印矩阵
4. [operator]ubuntu + git
5. gcc——预处理（预编译），编译，汇编，链接
6. [Luogu1040] 加分二叉树
7. python 机器移植
8. javaScript知识点大全
9. avalov+require实现tab栏
10. URAL 1993 This cheeseburger you don't need