需求：
1.从hive 表中获取数据。
2.计算各个指标与主分析指标间的相关系数。
3.将计算出来的相关系数，放入csv 文件，待使用。
首先，相关系数我们选择了皮尔逊相关系数，python的实现也是从网上直接找到的。
然后确定获取hive 数据的方式。公司环境没有pyspark，排除了用spark 操作hive 的方法。就想套用值之前脚本使用的，用impala 连接hive 的获取方法。结果基本开发完成，发现，一个是由于sql 数据量比较大，耗时比较长，导致总是自动断开连接，以至于跑不完程序。还有一个，由于其中的一个指标计算需要用到udf 函数，添加jar 包时，使用impala 的方式总是无法识别路径。后来认为应该是这种方式只能支持查询，无法支持这种添加临时函数的操作，没办法放弃了这一条路。
最后呢是选择了直接打开hive 的简单粗暴的模式：
os.popen("""hive -S -e '{}' """.format(sql))
ok 下面贴上我的代码，由于我是刚开始在工作中使用python，正在学习，肯定有很多不足的地方，如果能得到各位指点一二，那我真是非常感谢，不管是思路上的，还是代码上的，都希望大家不吝赐教。
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import datetime
import os
import sys
import pandas as pd
from math import sqrt
from operator import itemgetter, attrgetter
#执行hivesql，os.popen() 方法用于从一个命令打开一个管道，对于我来说正好是需要sql 的一行结果作为一个整体，刚好放到一个list 中。
#输出到指定日志文件
logging.basicConfig(level=logging.INFO,
filename='....../cor_coe.log',
filemode='a',
format='%(asctime)s %(filename)s %(levelname)s %(message)s',
datefmt='[%Y-%m-%d %H:%M:%S]'
)
def getHiveResult(sql):
output = os.popen("""hive -S -e '{}' """.format(sql))
result = output.readlines()
#形式：['字段名称\t字段名称','[第一行结果\t分隔]','[第二行结果\t分隔]'....]
return result
#这里是因为上面的result 这个list，每个元素的最后都加上了一个\n，我要把它去掉。
def trans_list(str):
return str.replace('\n','')
# 将'[...]' 这个字符串，转换成一个list.
def str2list(str):
list1 = str.split(',')
list1[0] = list1[0].replace('[','')
list1[-1] = list1[-1].replace(']','')
return [float(x) for x in list1]
#皮尔逊相关系数公式：x,y 两个变量的 协方差 / 标准差的乘积
# 乘积之和函数
def multipl(a,b):
sumofab=0.0
for i in range(len(a)):
temp=a[i]*b[i]
sumofab+=temp
return sumofab
# 皮尔逊相关系数函数
def corrcoef(x,y):
n=len(x)
#求和
sum1=sum(x)
sum2=sum(y)
#求乘积之和
sumofxy=multipl(x,y)
#求平方和
sumofx2 = sum([pow(i,2) for i in x])
sumofy2 = sum([pow(j,2) for j in y])
# 协方差 乘积之和 - 和的乘积
num=sumofxy-(float(sum1)*float(sum2)/n)
#标准差 * 标准差
den=sqrt((sumofx2-float(sum1**2)/n)*(sumofy2-float(sum2**2)/n))
return num/den
sql 就忽略了。sql 最终的结果形式是：
南京 [11,13,10.3,12.5..........] 这是一行记录，只有两个字段。
if __name__ == "__main__":
process_start = datetime.datetime.now()
logging.info("程序开始时间：" + str(process_start))
# 获取脚本外的日期参数,并添加连接符
ymd=sys.argv[1]
y_m_d="-".join((ymd[0:4],ymd[4:6],ymd[6:8]))
sql1 = sql1.format(DT=y_m_d)
try:
result = getHiveResult(sql1)
except Exception as e:
logging.info("调用getHiveResult()函数报错:" + traceback.format_exc())
list_result_name = ['','name1','name2','name3','name4']
list_result1 = result[1:]
list_result2 = []
list_result3 = []
for el1 in list_result1:
list_result2.append(trans_list(el1))
for el2 in list_result2:
list_tmp = []
list_tmp = el2.split('\t')
for el3 in list_tmp:
num = list_tmp.index(el3)
if num >=2:
try:
list_result3.append(( list_tmp[0],list_result_name[1]+"_"+list_result_name[num], round(corrcoef(str2list(list_tmp[1]),str2list(el3)),4) ))
except Exception as e:
logging.info("计算皮尔逊相关系数报错:" + traceback.format_exc())
#突然觉得python 的排序功能还挺强大的。在对我list 中的元祖排序
list_result3 = sorted(list_result3,key=itemgetter(0,2),reverse=True)
try:
df = pd.DataFrame(list_result3)
except Exception as e:
logging.info("list_result3转化成 df 时报错:" + traceback.format_exc())
outputpath = 'path/cor_coe_'+ymd+'.csv'
try:
df.to_csv(outputpath,index=False,sep=',',header=['城市','相关因子','相关系数'])
except Exception as e:
logging.info("输出到 csv 文件报错:" + traceback.format_exc())
print ('process_end:',datetime.datetime.now())
现在我还有一个瓶颈，就是sql 跑的太慢了，有十个sql，每个sql 涉及到的表都在千万到亿条数据左右，总共的时间需要22min左右，脚本的总时间是25min左右。已经尝试设置了很多hive 所谓的优化参数，除了合并小文件，提升了将近4分钟的速度之外，其他的调整都没有什么效果。而且从执行过程来看，并没有明显的数据倾斜，就是map 和 reduce 的过程就比较慢。如果后期找到方法之后我还会追加进来。也希望看到这篇文章的朋友能提一些宝贵的意见。
网友评论