在Python中使用ibis,impyla,pyhive,pyspark连接Kerberos安全认证的Hive、Impala

本是古典何须时尚 2023-02-15 05:25 9阅读 0赞

在python中连接hive和impala有很多中方式，有**pyhive,impyla,pyspark,ibis**等等，本篇我们就逐一介绍如何使用这些包连接hive或impala，以及如何通过kerberos认证。

# Kerberos #

如果集群没开启kerberos认证则不需要这里的代码，或者在系统环境内通过kinit命令认证也不需要这部分的代码。

**krbcontext.context\_shell**

# -*- coding: utf-8 -*-
    
    __all__ = [ 'krbcontext', 'KRB5KinitError', ]
    
    import os, sys
    # import pwd
    import subprocess
    
    
    from contextlib import contextmanager
    
    class KRB5KinitError(Exception):
        pass
    
    # def get_login():
    #     ''' Get current effective user name '''
    #
    #     return pwd.getpwuid(os.getuid()).pw_name
    
    def init_ccache_as_regular_user(principal=None, ccache_file=None):
        '''Initialize credential cache as a regular user
    
        Return the filename of newly initialized credential cache
        '''
    
        if not sys.stdin.isatty():
            raise IOError('This is not running on console. So, you need to run kinit '
                          'with your principal manually before anything goes.')
    
        cmd = 'kinit %(ccache_file)s %(principal)s'
        args = {}
    
        args['principal'] = principal
        args['ccache_file'] = '-c %s' % ccache_file
    
        kinit_proc = subprocess.Popen(
            (cmd % args).split(),
            stderr=subprocess.PIPE)
        stdout_data, stderr_data = kinit_proc.communicate()
    
        if kinit_proc.returncode > 0:
            raise KRB5KinitError(stderr_data)
    
        return ccache_file
    
    
    
    def init_ccache_with_keytab(principal, keytab_file, ccache_file):
        '''Initialize credential cache using keytab file
    
        Return the filename of newly initialized credential cache
        '''
        cmd = 'kinit -kt %(keytab_file)s -c %(ccache_file)s %(principal)s'
        args = {}
    
        args['principal'] = principal
        args['ccache_file'] = ccache_file
        args['keytab_file'] = keytab_file
    
        kinit_proc = subprocess.Popen(
            (cmd % args).split(),
            stderr=subprocess.PIPE)
        stdout_data, stderr_data = kinit_proc.communicate()
    
        if kinit_proc.returncode > 0:
            raise KRB5KinitError(stderr_data)
    
        return ccache_file
    
    
    @contextmanager
    def krbcontext(using_keytab=False, **kwargs):
        '''A context manager for Kerberos-related actions
    
        using_keytab: specify to use Keytab file in Kerberos context if True,
                      or be as a regular user.
        kwargs: contains the necessary arguments used in kerberos context.
                It can contain principal, keytab_file, ccache_file.
                When you want to use Keytab file, keytab_file must be included.
        '''
    
        env_name = 'KRB5CCNAME'
        old_ccache = os.getenv(env_name)
        if using_keytab:
            ccache_file = init_ccache_with_keytab(**kwargs)
        else:
            ccache_file = init_ccache_as_regular_user(kwargs.get("principal"), kwargs.get("ccache_file"))
        os.environ[env_name] = ccache_file
        yield

# pyhive #

使用pyhive连接hive

### 环境 ###

"""
     decorator==4.4.2
     future==0.18.2
     gssapi==1.6.5
     krbcontext==0.10
     PyHive==0.5.0
     impyla==0.14.1
     sasl==0.2.1
     six==1.11.0
     thrift_sasl==0.3.0  # 如果使用ibis或impyla需要thrift_sasl==0.2.1
     thrift==0.13.0
     thriftpy==0.3.5
     """
    
    from pyhive import sqlalchemy_hive,hive
    from krbcontext.context_shell import krbcontext

### Kerberos认证 ###

两种方式，一种是在系统环境内让当前用户通过kinit 先保持kerberos的认证，那所有的代码里就无需再写kerberos认证代码了。另一种方式，在python脚本内使用下面的代码进入kerberos认证会话：

config = {
        "kerberos_principal": "hive@CSDNTEST.COM.LOCAL",
        "keytab_file": '/home/tools/wyk/keytab/hive.keytab',  
        "kerberos_ccache_file": '/home/tools/wyk/keytab/hive_ccache_uid',
        "AUTH_MECHANISM": "GSSAPI"
    }
    with krbcontext(using_keytab=True,
                                   principal=config['kerberos_principal'],
                                   keytab_file=config['keytab_file'],
                                   ccache_file=config['kerberos_ccache_file']):
        #在这个代码块内执行的脚本都是在kerberos认证内的。
        #可在此代码块内查询hdfs hive hbase等等只要是kerberos principal有权限的组件都可以

### 使用 ###

下面的代码需要在上面的kerberos代码块内才能通过kerberos认证，如果集群没有开启kerberos认证或当期系统已使用kinit进行了认证则无需上面的代码：

con = hive.connect(host='uatnd02.csdntest.com.local',port=10000,auth='KERBEROS',kerberos_service_name="hive") #host为hiveserver2所在节点，port默认10000，为hs2的端口
        cursor = con.cursor()
        cursor.execute('select * from dl_nccp.account limit 5') #不能有分号！
        # cursor.execute('desc dl_nccp.account') #不能有分号！
        datas = cursor.fetchall()
        print(datas)
        cursor.close()
        con.close()

# impyla #

### 环境 ###

"""
     decorator==4.4.2
     future==0.18.2
     gssapi==1.6.5
     krbcontext==0.10
     PyHive==0.5.0
     impyla==0.14.1
     sasl==0.2.1
     six==1.11.0
     thrift_sasl==0.2.1  # 只有这个与pyhive的不一样
     thrift==0.13.0
     thriftpy==0.3.5
     """
    
    from impala.dbapi import connect
    from krbcontext.context_shell import krbcontext

### Kerberos认证 ###

与上面的pyhive一样，略

### 使用 ###

#impyla  thrift==0.2.1  与ibis依赖版本相同，可同时使用
        conn = connect(host='uatnd02.csdntest.com.local', port=10000, auth_mechanism='GSSAPI',kerberos_service_name='hive')
        cur = conn.cursor()
        cur.execute('SHOW databases') #不能有分号
        cur.table_exists(table_name='account',database_name='dl_nccp') #return True or False
        cur.ping() #return True or False
        cur.status() #return True or False
        cur.get_table_schema(table_name='account',database_name='dl_nccp') #return 表结构 类似desc
        print(cur.fetchall())
        cur.close()
        conn.close()

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGMwNTIx_size_16_color_FFFFFF_t_70][]

# ibis #

ibis是个很强大的第三方环境包，支持对各类数据库或文件系统进行访问，功能强大，并返回pandas dataframe格式，对数据开发人员很友好，这个包我用了两年多，非常**推荐**。

官方文档：[https://docs.ibis-project.org/getting-started.html][https_docs.ibis-project.org_getting-started.html]

### 环境 ###

"""
     ibis-framework==0.14.0  #pip install ibis-framework[impala]==0.14.0
     decorator==4.4.2
     future==0.18.2
     gssapi==1.6.5
     krbcontext==0.10
     PyHive==0.5.0
     impyla==0.14.1
     sasl==0.2.1
     six==1.11.0
     thrift_sasl==0.2.1  # 只有这个与pyhive的不一样
     thrift==0.13.0
     thriftpy==0.3.5
     """

### Kerberos认证 ###

与上面的pyhive一样，略

### 使用 ###

官方文档以及源码里还有很多函数可供使用，这里仅列出常用的一些，建议使用ibis包的可以看看源码或官方文档。

import ibis
    import pandas as pd
    import ibis.expr.datatypes as dt
    conf={
    "impala_host":"uathd01.csdntest.com.local",
    "impala_port":21050,
    "kerberos_service_name":"impala",
    "auth_mechanism":"GSSAPI",
    "webhdfs_host1":"uatnd01.csdntest.com.local",
    "webhdfs_host2":"uatnd02.csdntest.com.local",
    "webhdfs_port":9870   #如果是hadoop3之前的版本这里的参数改为50070
    }
    #获取hdfs连接
    try:
        hdfs_client = ibis.hdfs_connect(host = conf["webhdfs_host2"], port = conf["webhdfs_port"], auth_mechanism = conf["auth_mechanism"], use_https = False, verify = False)
        hdfs_client.ls('/')
    except:
        hdfs_client = ibis.hdfs_connect(host = conf["webhdfs_host1"], port = conf["webhdfs_port"], auth_mechanism = conf["auth_mechanism"], use_https = False, verify = False)
    
    #获取impala连接
    impala_client = ibis.impala.connect(host=conf["impala_host"], port=conf["impala_port"], hdfs_client = hdfs_client, auth_mechanism=conf["auth_mechanism"], timeout = 300)
    
    # 直接读表，select * from dh_sales.r_order limit 10，返回pandas dataframe
    res = impala_client.table('r_order', database='dh_sales').execute(limit=10)
    print(type(res))
    print(res.dtypes)
    print(res)
    
    #使用SQL查询impala
    res = impala_client.sql("""select area,is_oversea,dw_ins_ts,sum(gid_cnt) from dh_t3_report_crm.r_duration_area_metr where is_oversea=0 group by 1,2,3 limit 5""")
    df_res = res.execute()
    print(df_res)
    
    #列库下所有表
    impala_client.list_tables(database='dl_common')
    
    #使用pandas dataframe建表，类型自动映射
    impala_client.create_table(table_name='ibis_create0602',obj=sys_dict,database='default',force=True)
    ...
    #pd_res_acc 略
    #pd_res_cre 略
    ...
    #使用pandas datraframe插入数据
    impala_client.insert(table_name='tmp_account',obj=pd_res_acc,database='default',overwrite=True)
    impala_client.insert(table_name='ibis_create',obj=pd_res_cre,database='default',overwrite=True)

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGMwNTIx_size_16_color_FFFFFF_t_70 1][]

![20200609114416317.png][]

![20200609114625752.png][]

### BUG解决 ###

执行insert或create命令可能会报下面的错误，这个是源码级别的BUG，参考下面的连接修改源码即可解决：

UnicodeEncodeError: 'latin-1' codec can't encode characters in position 160-161: ordinal not in range(256)

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGMwNTIx_size_16_color_FFFFFF_t_70 2][]

参考：[https://github.com/ibis-project/ibis/issues/2120][https_github.com_ibis-project_ibis_issues_2120]

*vim /home/tools/python3/**Python-3.6.8/lib/python3.6/site-packages/hdfs/client.py ***

![202006091132096.png][]

# pyspark #

### 环境 ###

*vim /etc/profile*

export SPARK_HOME=/opt/cloudera/parcels/CDH/lib/spark
    export SPARK_CONF_DIR=$SPARK_HOME/conf
    export PYTHONPATH=/home/tools/anaconda3/envs/csdn/bin
    export PYSPARK_PYTHON=/home/tools/anaconda3/envs/csdn/bin/python3.6

"""
    pyspark==2.4.5
    """

### kerberos认证 ###

与上面的pyhive一样，略

### 使用 ###

使用pyspark连接hive进行查询，并将spark dataframe转为pandas dataframe：

from __future__ import division
    #import findspark as fs
    #fs.init()
    import pandas as pd
    from pyspark.sql import HiveContext,SparkSession,SQLContext
    from pyspark import SparkContext, SparkConf
    import pyspark.sql.functions as F
    import datetime as dt
    from datetime import datetime
    import random
    import numpy as np
    from log3 import log_to_file, log
    from pyspark.sql.types import *
    
    conf = SparkConf().setMaster("yarn").setAppName("MF_return_calc")
    sc = SparkContext(conf=conf)
    sc.setLogLevel("WARN")
    hiveCtx = HiveContext(sc)
    spark = SparkSession.builder.master("yarn").appName("MF_return_calc").config("spark.debug.maxToStringFields", "100").getOrCreate()
    
    #执行SQL
    test_sql = """select * from dl_nccp.account limit 5"""
    res = hiveCtx.sql(test_sql)
    type(res) #返回spark dataframe
    res.head(3)
    res_pd = res.toPandas() #将spark dataframe转为pandas dataframe
    res_pd

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGMwNTIx_size_16_color_FFFFFF_t_70 3][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGMwNTIx_size_16_color_FFFFFF_t_70 4][]

好了上面四种在python中访问hive和impala的方式就介绍完了，希望对大家有帮助，谢谢！

希望本文对你有帮助，请点个赞鼓励一下作者吧~ 谢谢！

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGMwNTIx_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200609111326452.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGMwNTIx,size_16,color_FFFFFF,t_70
[https_docs.ibis-project.org_getting-started.html]: https://docs.ibis-project.org/getting-started.html
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGMwNTIx_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/2020060911415873.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGMwNTIx,size_16,color_FFFFFF,t_70
[20200609114416317.png]: https://img-blog.csdnimg.cn/20200609114416317.png
[20200609114625752.png]: https://img-blog.csdnimg.cn/20200609114625752.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGMwNTIx_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20200609112458435.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGMwNTIx,size_16,color_FFFFFF,t_70
[https_github.com_ibis-project_ibis_issues_2120]: https://github.com/ibis-project/ibis/issues/2120
[202006091132096.png]: https://img-blog.csdnimg.cn/202006091132096.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGMwNTIx_size_16_color_FFFFFF_t_70 3]: https://img-blog.csdnimg.cn/20200609131634995.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGMwNTIx,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGMwNTIx_size_16_color_FFFFFF_t_70 4]: https://img-blog.csdnimg.cn/20200609131821212.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGMwNTIx,size_16,color_FFFFFF,t_70