使用python从蜂巢读取数据时出现性能问题 [英] Performance issue while reading data from hive using python
问题描述
我在蜂巢中有一个表，其中包含351837条记录(110 MB大小)，我正在使用python读取此表并写入sql服务器.
I have a table in hive with 351 837 (110 MB size) records and i am reading this table using python and writing into sql server.
在此过程中，将数据从蜂巢读取到熊猫数据帧时需要花费很长时间.当我加载全部记录(351k)时，需要90分钟.
In this process while reading data from hive into pandas dataframe it is taking long time. When i load entire records (351k) it takes 90 minutes.
为了改进，我采用了以下方法，例如从蜂巢中读取1万行并写入sql server.但是，仅从配置单元读取1万行并将其分配给Dataframe仅需要4到5分钟的时间.
To improve i went with following approach like reading 10k rows once from hive and writing into sql server. But reading 10k rows once from hive and assigning it to Dataframe is alone taking 4-5 minutes of time.
def execute_hadoop_export():
"""
This will run the steps required for a Hadoop Export.
Return Values is boolean for success fail
"""
try:
hql='select * from db.table '
# Open Hive ODBC Connection
src_conn = pyodbc.connect("DSN=****",autocommit=True)
cursor=src_conn.cursor()
#tgt_conn = pyodbc.connect(target_connection)
# Using SQLAlchemy to dynamically generate query and leverage dataframe.to_sql to write to sql server...
sql_conn_url = urllib.quote_plus('DRIVER={ODBC Driver 13 for SQL Server};SERVER=Xyz;DATABASE=Db2;UID=ee;PWD=*****')
sql_conn_str = "mssql+pyodbc:///?odbc_connect={0}".format(sql_conn_url)
engine = sqlalchemy.create_engine(sql_conn_str)
# read source table.
vstart=datetime.datetime.now()
for df in pandas.read_sql(hql, src_conn,chunksize=10000):
vfinish=datetime.datetime.now()
print 'Finished 10k rows reading from hive and it took', (vfinish-vstart).seconds/60.0,' minutes'
# Get connection string for target from Ctrl.Connnection
df.to_sql(name='table', schema='dbo', con=engine, chunksize=10000, if_exists="append", index=False)
print 'Finished 10k rows writing into sql server and it took', (datetime.datetime.now()-vfinish).seconds/60.0, ' minutes'
vstart=datetime.datetime.now()
cursor.Close()
except Exception, e:
print str(e)
输出:
在python中读取配置单元表数据最快的方法是什么?
What is the fastest way to read hive table data in python?
更新配置单元表结构
CREATE TABLE `table1`(
`policynumber` varchar(15),
`unitidentifier` int,
`unitvin` varchar(150),
`unitdescription` varchar(100),
`unitmodelyear` varchar(4),
`unitpremium` decimal(18,2),
`garagelocation` varchar(150),
`garagestate` varchar(50),
`bodilyinjuryoccurrence` decimal(18,2),
`bodilyinjuryaggregate` decimal(18,2),
`bodilyinjurypremium` decimal(18,2),
`propertydamagelimits` decimal(18,2),
`propertydamagepremium` decimal(18,2),
`medicallimits` decimal(18,2),
`medicalpremium` decimal(18,2),
`uninsuredmotoristoccurrence` decimal(18,2),
`uninsuredmotoristaggregate` decimal(18,2),
`uninsuredmotoristpremium` decimal(18,2),
`underinsuredmotoristoccurrence` decimal(18,2),
`underinsuredmotoristaggregate` decimal(18,2),
`underinsuredmotoristpremium` decimal(18,2),
`umpdoccurrence` decimal(18,2),
`umpddeductible` decimal(18,2),
`umpdpremium` decimal(18,2),
`comprehensivedeductible` decimal(18,2),
`comprehensivepremium` decimal(18,2),
`collisiondeductible` decimal(18,2),
`collisionpremium` decimal(18,2),
`emergencyroadservicepremium` decimal(18,2),
`autohomecredit` tinyint,
`lossfreecredit` tinyint,
`multipleautopoliciescredit` tinyint,
`hybridcredit` tinyint,
`goodstudentcredit` tinyint,
`multipleautocredit` tinyint,
`fortyfivepluscredit` tinyint,
`passiverestraintcredit` tinyint,
`defensivedrivercredit` tinyint,
`antitheftcredit` tinyint,
`antilockbrakescredit` tinyint,
`perkcredit` tinyint,
`plantype` varchar(100),
`costnew` decimal(18,2),
`isnocontinuousinsurancesurcharge` tinyint)
CLUSTERED BY (
policynumber,
unitidentifier)
INTO 50 BUCKETS
注意:我也尝试了sqoop导出选项，但是我的配置单元表已经处于存储桶格式.
Note: I have also tried with sqoop export option but my hive table is already in bucketting format.
推荐答案
我尝试了多处理，我可以将其从2小时减少8-10分钟.请找到以下脚本.
I have tried with multi-processing and i can reduce it 8-10 minutes from 2 hours. Please find below scripts.
from multiprocessing import Pool
import pandas as pd
import datetime
from query import hivetable
from write_tosql import write_to_sql
p = Pool(37)
lst=[]
#we have 351k rows so generating series to use in hivetable method
for i in range(1,360000,10000):
lst.append(i)
print 'started reading ',datetime.datetime.now()
#we have 40 cores in cluster
p = Pool(37)
s=p.map(hivetable, [i for i in lst])
s_df=pd.concat(s)
print 'finished reading ',datetime.datetime.now()
print 'Started writing to sql server ',datetime.datetime.now()
write_to_sql(s_df)
print 'Finished writing to sql server ',datetime.datetime.now()
--------- query.py文件-------
---------query.py file-------
import pyodbc
from multiprocessing import Pool
from functools import partial
import pandas as pd
conn = pyodbc.connect("DSN=******",autocommit=True)
def hivetable(row):
query = 'select * from (select row_number() OVER (order by policynumber) as rownum, * from dbg.tble ) tbl1 where rownum between '+str(row) +' and '+str(row+9999)+';'
result = pd.read_sql(query,conn)
return result
--------- Write_tosql.py文件---------
---------Write_tosql.py file---------
import sqlalchemy
import urllib
import pyodbc
def write_to_sql(s_df):
sql_conn_url = urllib.quote_plus('DRIVER={ODBC Driver 13 for SQL Server};SERVER=ser;DATABASE=db;UID=sqoop;PWD=#####;')
sql_conn_str = "mssql+pyodbc:///?odbc_connect={0}".format(sql_conn_url)
engine = sqlalchemy.create_engine(sql_conn_str)
s_df.rename(columns=lambda x: remove_table_alias(x), inplace=True)
s_df.to_sql(name='tbl2', schema='dbo', con=engine, chunksize=10000, if_exists="append", index=False)
def remove_table_alias(columnName):
try:
if(columnName.find(".") != -1):
return columnName.split(".")[1]
return columnName
except Exception, e:
print "ERROR in _remove_table_alias ",str(e)
任何其他解决方案都可以帮助我减少时间.
Any other solutions will help me to reduce in time.
这篇关于使用python从蜂巢读取数据时出现性能问题的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！
