python phoenix API

电玩女神 2022-01-11 05:21 289阅读 0赞

### **一、phoenix** ###

Phonix是搜索引擎，n个版本之前是作为独立的产品存在的，现在集成到hbase里面。Phoenix是一个开源的HBASE SQL层。Phoeinx可以用标准的JDBC API替代HBASE client API来创建表，插入和查询查询HBASE中的数据。

**Phoenix作为应用层和HBASE之间的中间件,以下特性使它在大数据量的简单查询场景有着独有的优势**

*   二级索引支持(global index + local index)
 *  编译SQL成为原生HBASE的可并行执行的scan
 *  在数据层完成计算，server端的coprocessor执行聚合
 *  下推where过滤条件到server端的scan filter上
 *  利用统计信息优化、选择查询计划（5.x版本将支持CBO）
 *  skip scan功能提高扫描速度

**一句话：用sql语句操作no sql数据**

### ** 二、接口调用** ###

phoenix有phoenixdb 和JayDeBeApi两个包，这里采用的是phoenixdb。好像没有连接池方式、只能实时打开、关闭或者长链接方式。

下载依赖包 pip3 install phoenixdb

*  configuration.properties

[Hbase]
    hbase_host:172.8.10.xx
    hbase_db:xx

*  PhoenixClient

#!/usr/bin/python3
    # -*- coding: UTF-8 -*-
    
    import json
    import pandas as pd
    import phoenixdb as pdb
    
    ## READ CONFIGURATION FILE
    config_file = pd.read_table(filepath_or_buffer="configuration.properties",
                                header=None, delim_whitespace=True,index_col=0).transpose()
    
    config=ConfigParser()
    config.read(str(config_file['configPath'].iloc[0]))
    
    # =============================================================================
    # initialization
    # =============================================================================
    
    Hbase_host = config.get('Hbase', 'Hbase_host')
    Hbase_db = config.get('Hbase', 'Hbase_db')
    
    
    class PhoenixClient(object):
    
        def __init__(self):
            self.url = 'http://{}:8765/'.format(Hbase_host)
            self.Hbase_db = Hbase_db
    
        def pullData(self,query_str,ID=False):
            try:
                self.conn=pdb.connect('http://{}:8765/'.format(Hbase_host), autocommit=True)
                self.cursor = self.conn.cursor()
                self.cursor.execute(query_str)
                cols = [x.name for x in self.cursor.description]
                data = self.cursor.fetchall()
                if ID:
                    return pd.DataFrame(data,columns=cols)
                else:
                    return pd.DataFrame(data,columns=cols).drop(columns="ID")
            finally:
                self.cursor.close()
                self.conn.close()
    
    
        def pushData(self,table,df,rowKeyName,prefix):
            try:
                self.conn=pdb.connect('http://{}:8765/'.format(Hbase_host), autocommit=True)
                self.cursor = self.conn.cursor()
                ## add ID column as ROWKEY
                df = df.assign(ID = df[rowKeyName].apply(lambda x:'_'.join(x.map(str)),axis=1))
                df = df.assign(ID = prefix+'_'+df['ID']+'_'+df.index.map(str))
                cols = tuple(df.columns.tolist())
                hbTable = '{}:{}'.format(self.Hbase_db,table)
                query_str = '''UPSERT INTO "{}"{} VALUES ({})'''.format(hbTable,cols,','.join(['?']*len(cols)))
                query_str=query_str.replace("'","\"")
                self.cursor.executemany(query_str, df.to_numpy().tolist())
            finally:
                self.cursor.close()
                self.conn.close()
    
    
        def executer(self,query_str):
            try:
                self.conn=pdb.connect('http://{}:8765/'.format(Hbase_host), autocommit=True)
                self.cursor = self.conn.cursor()
                self.cursor.execute(query_str)
            finally:
                self.cursor.close()
                self.conn.close()