PyHive 0.7.0

DB-API

from pyhive import presto  # or import hive or import trino
cursor = presto.connect('localhost').cursor()  # or use hive.connect or use trino.connect
cursor.execute('SELECT * FROM my_awesome_data LIMIT 10')
print cursor.fetchone()
print cursor.fetchall()

DB-API（异步）

from pyhive import hive
from TCLIService.ttypes import TOperationState
cursor = hive.connect('localhost').cursor()
cursor.execute('SELECT * FROM my_awesome_data 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

    # If needed, an asynchronous query can be cancelled at any time with:
    # cursor.cancel()

    status = cursor.poll().operationState

print cursor.fetchall()

在Python 3.7中 async 成为了一个关键字；你可以使用 async_ 代替

cursor.execute('SELECT * FROM my_awesome_data LIMIT 10', async_=True)

SQLAlchemy

首先安装此包，并将其与SQLAlchemy注册，参见 setup.py 中的 entry_points。

from sqlalchemy import *
from sqlalchemy.engine import create_engine
from sqlalchemy.schema import *
# Presto
engine = create_engine('presto://localhost:8080/hive/default')
# Trino
engine = create_engine('trino+pyhive://localhost:8080/hive/default')
# Hive
engine = create_engine('hive://localhost:10000/default')

# SQLAlchemy < 2.0
logs = Table('my_awesome_data', MetaData(bind=engine), autoload=True)
print select([func.count('*')], from_obj=logs).scalar()

# Hive + HTTPS + LDAP or basic Auth
engine = create_engine('hive+https://username:password@localhost:10000/')
logs = Table('my_awesome_data', MetaData(bind=engine), autoload=True)
print select([func.count('*')], from_obj=logs).scalar()

# SQLAlchemy >= 2.0
metadata_obj = MetaData()
books = Table("books", metadata_obj, Column("id", Integer), Column("title", String), Column("primary_author", String))
metadata_obj.create_all(engine)
inspector = inspect(engine)
inspector.get_columns('books')

with engine.connect() as con:
    data = [{ "id": 1, "title": "The Hobbit", "primary_author": "Tolkien" },
            { "id": 2, "title": "The Silmarillion", "primary_author": "Tolkien" }]
    con.execute(books.insert(), data[0])
    result = con.execute(text("select * from books"))
    print(result.fetchall())

注意：查询生成功能可能并不完整或经过全面测试，但应该不会对原始SQL造成问题。

传递会话配置

# DB-API
hive.connect('localhost', configuration={'hive.exec.reducers.max': '123'})
presto.connect('localhost', session_props={'query_max_run_time': '1234m'})
trino.connect('localhost',  session_props={'query_max_run_time': '1234m'})
# SQLAlchemy
create_engine(
    'presto://user@host:443/hive',
    connect_args={'protocol': 'https',
                  'session_props': {'query_max_run_time': '1234m'}}
)
create_engine(
    'trino+pyhive://user@host:443/hive',
    connect_args={'protocol': 'https',
                  'session_props': {'query_max_run_time': '1234m'}}
)
create_engine(
    'hive://user@host:10000/database',
    connect_args={'configuration': {'hive.exec.reducers.max': '123'}},
)
# SQLAlchemy with LDAP
create_engine(
    'hive://user:password@host:10000/database',
    connect_args={'auth': 'LDAP'},
)