使用python连接hive数据库并进行数据操作

在数据抽取或者数据存取过程中难免会遇到用其他语言对hive数据库进行操作。如python远程对hive数据库进行操作，需要通过thrift服务进行操作。本文的环境是python==3.6.1， hive==1.1.0，thfit==0.10.0

ThriftServer介绍

客户端对hive数据库进行操作可以通过Cli进行，在本地这种方式较好，但是在远程操作时会很麻烦，因此提供了hiveserver和hiveserver2，在不启动cli的情况下对hive进行操作，可以在远程通过其他语言(java, python, php等)向hive请求并返回结果。也就是利用thrift进行整个过程操作。

hiveserver2主要是向远程调用提供了接口，通过thrift rpc实现，进行远程的操作。可以实现远程并行操作hive数据库。启动hiveserver2如下：

1	nohup hive --service hiveserver2 &

在启动后hiveserver2默认监听的端口为10000，可以在hive的配置文件(hive-site.xml)中查看或者修改该端口。可以通过命令查看该端口是否被监听：netstat -antp | grep 10000

连接hive的框架

在连接hive时，可以选择thrift本身，pyhive，pyhs2和impyla。在使用过程中首先尝试了thrift本身，但是通过配置后在连接返回：thrift.transport.TTransport.TTransportException: None，据网友说这是连接hiveserver2出现的问题。因此弃用thrift直接连接hive，选择其他三个。

pyhs2是以前hive官方推荐使用的库，主要依赖了thrift和sasl。但是这个库后面没有维护了，因此在最新的python和hive下有很多问题，因此弃用。

impyla是通过impala来对操作hive，目前使用的环境中impala没有启用，因此该库就放弃使用了。因此只剩下的pyhive。但是网上的使用者反应，推荐使用pyhive和impyla。

配置pyhive的运行环境

因为通过比较选择了pyhive和hive进行交互，因此需要在客户端部署pyhive，不需要在服务端安装pyhive。本文中主要是针对linux系统上的部署，不考虑windows上的部署。

安装依赖包sasl的环境. 当安装了下面的sasl相关的部署包才能正确安装sasl

# ubuntu
sudo apt-get install sasl2-bin libsasl2-2 libsasl2-dev libsasl2-modules
# centos
sudo yum install cyrus-sasl-devel cyrus-sasl-gssapi cyrus-sasl-md5 cyrus-sasl-plain
# 使用pip安装python下的sasl
pip install sasl==0.2.1

安装thrift的python包：pip install thrift==0.10.0
安装thrift_sasl，者个包依赖了sasl和thrift：pip install thrift_sasl====0.3.0
如果是ubuntu，则会多出一个步骤，否则在引用sasl包时可能会报错：_ZTVNSt7__cxx1118basic_stringstreamIcSt11char_traitsIcESaIcEEE，因此需要对python使用的libgcc进行更新，因为我部署的python环境是anaconda3因此直接执行conda install libgcc即可。
安装pyhive，pyhive的安装很简单：pip install pyhive==0.5.0

经过上面的安装可以进行包的测试：

1
2
3

python -c "import sasl"
python -c "import thrift"
python -c "import pyhive"

测试成功后就可以连接服务端操作hive：

from pyhive import hive
from TCLIService.ttypes import TOperationState
cursor = hive.connect('localhost').cursor()
cursor.execute('SELECT * FROM table LIMIT 10', async=True)

status = cursor.poll().operationState
while status in (TOperationState.INITIALIZED_STATE, TOperationState.RUNNING_STATE):
    logs = cursor.fetch_logs()
    for message in logs:
        print message
    status = cursor.poll().operationState

print cursor.fetchall()