数据库
首页 > 数据库> > hive导致表锁的SQL和租户信息获取

hive导致表锁的SQL和租户信息获取

作者:互联网

hive导致表锁的SQL和租户信息获取

多租户平台存在同一个表有多个租户具有访问权限,hive提供锁机制,保障数据的安全。由此会引发由于某个租户占用锁时间较长,其他租户作业滞后,如果将导致表锁的SQL和租户信息截取到,可以提供给局方进行业务流程的优化,加快作业执行效率。本文档记录一个可实现的方法供参考

获取表锁信息

场景:hive使用分布式协调服务(Zookeeper)提供的分布式锁,hive底层由mapreduce执行,yarn统一资源调度,本文档由python语言实现

关键代码

# xxxx
# 2021/11/4 8:54
import configparser
import os
import time
from kazoo.client import KazooClient
import ConnMysqlTmpl

'''
#获取zookeeper集群信息
ZookeeperInfo.ini文件为:
[zookeeper-conn]
url=zk_host_name:port
'''
def zookeeper_info():
   cf = configparser.ConfigParser()
   pass1 = os.path.dirname(os.path.abspath('.'))
   cf.read(pass1 + "/config/ZookeeperInfo.ini")
   items = cf.items("zookeeper-conn")
   list = []
   for line in items:
       list.append(line[1])
   return list

'''
#定义一个zookeeper类,实现zk集群的链接和关闭
'''
class Zookeeper:
   def connection(self,zk_url):
       self.zk_url=zk_url
       try:
           self.conn_path=KazooClient(self.zk_url)
           self.conn_path.start()
           return self.conn_path
       except Exception as e:
           print(f"SSH连接异常, 错误如下: {e}")

   def conn_close(self):
       self.conn_path.stop()
       print("已关闭Zookeeper连接...")

'''
#获取需要监控的表
HiveMonLockInfo文件格式随意,只要在zk中拿到需要监控的表路径即可
'''
def monitor_info():
   filename = 'datainfo/HiveMonLockInfo'
   if os.path.exists(filename):
       with open(filename, 'r', encoding='utf8') as rfile:
           lines = rfile.readlines()
       return lines
   else:
       print("未找到文件:【 {0} 】信息,请联系后台管理员处理".format(filename))

'''
#获取表锁列表
'''
def get_lock():
   zk=Zookeeper()
   zk_info = zookeeper_info()
   zk_conn=zk.connection(zk_info[0])
   table_info=monitor_info()
   locks = []
   for table in table_info:
       if zk_conn.exists("/hive_zookeeper_namespace/" + table.rstrip('\n')):
           lock_name = zk_conn.get_children("/hive_zookeeper_namespace/" + table.rstrip('\n'))
           for lock in lock_name:
               if "LOCK" in lock:
                   locks.append("/hive_zookeeper_namespace/"+table.rstrip('\n')+"/"+lock)
               else:
                   pass
       else:
           print("/hive_zookeeper_namespace/" + table.rstrip('\n'), "无锁情况", get_time())
   zk.conn_close()
   return locks

'''
#获取每个表锁的具体信息
table_locks_info为入库执行的语句
'''
def get_lock_info():
   zk = Zookeeper()
   zk_info = zookeeper_info()
   zk_conn = zk.connection(zk_info[0])
   all_locks = get_lock()
   print(all_locks)
   if all_locks:
       for path in all_locks:
           #获取所有锁node信息
           all_info=zk_conn.get(path)

           #获取query_id和sql语句
           all_sql_info = str(all_info).split("ZnodeStat")[0]
           query_id = (all_sql_info.split(":")[0])[3:]
           sql_info_tmp = all_sql_info.split(":")[3]
           sql_info_tmp1=sql_info_tmp.replace("\n","\\n")
           sql_info=sql_info_tmp1.replace("\'","\\'")

           #获取节点创建信息
           time_info = str(all_info).split("ZnodeStat")[1]
           create_time = time_info.split(",")[2]

           #时间戳转换
           timeStamp = int(create_time.split("=")[1])
           '''
          爬取下来的时间戳长度都是13位的数字,而time.localtime的参数要的长度是10位,所以我们需要将其/1000并取整即可
          '''
           timeArray = time.localtime(timeStamp/1000)
           otherStyleTime = time.strftime("%Y-%m-%d %H:%M:%S", timeArray)
           try:
               ConnMysqlTmpl.table_locks_info(query_id,sql_info,otherStyleTime,path.split("/")[3])
           except Exception as e:
               print(f"执行SQL失败,请排查:{e}")
   else:
       print("所监控的表无锁情况~", get_time())
   zk.conn_close()

def get_time():
   create_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
   return create_time

if __name__ == '__main__':
   get_lock_info()

主要说明

#获取所有锁node信息

#获取query_id和sql语句
all_sql_info = str(all_info).split("ZnodeStat")[0]
query_id = (all_sql_info.split(":")[0])[3:]
sql_info_tmp = all_sql_info.split(":")[3]
sql_info_tmp1=sql_info_tmp.replace("\n","\\n")
sql_info=sql_info_tmp1.replace("\'","\\'")

/SQL语句和SQL语句的query_id/

获取运行作业的query_id和租户

关键代码

import requests
import configparser
import os
import AppSourceInfo
import ConnMysqlTmpl
from json import JSONDecodeError
from requests.auth import HTTPBasicAuth

'''
#获取配置信息:访问url
'''
def get_app_info():
   cf=configparser.ConfigParser()
   pass1=os.path.dirname(os.path.abspath("."))
   cf.read(pass1 + "/config/AppExecSqlInfo.ini")
   items = cf.items("appexecsql-conn")
   list = []
   for line in items:
       list.append(line[1])
   return list

'''
#获取运行状态下作业的query_id和租户
'''
def job_info():
   conn_info=get_app_info()
   app_ids = AppSourceInfo.get_each_app1()
   if app_ids:
       for line in app_ids:
           try:
               all_info_tmp=requests.get(conn_info[0]+line+"/ws/v1/mapreduce/jobs/"+str(line).replace("application","job")+"/conf",
                             auth=HTTPBasicAuth(conn_info[1],conn_info[2]))
               try:
                   info_json=all_info_tmp.json()
                   result=info_json["conf"]
                   all_info=result["property"]
                   list=[]
                   for line in all_info:
                       '''
                      if line["name"] == "hive.query.string":
                          #list.append(str(line["value"]))
                          sql_info_tmp = line["value"]
                          if sql_info_tmp:
                              sql_info_tmp1 = sql_info_tmp.replace("\n", "\\n")
                              sql_info = sql_info_tmp1.replace("\'", "\\'")
                              list.append(sql_info)
                          else:
                              sql_info = "0"
                              list.append(sql_info)
                      elif line["name"] == "mapreduce.jdbc.input.query":
                          #list.append(str(line["value"]))
                          jdbc_query_tmp = line["value"]
                          #print(type(jdbc_query_tmp))
                          if len(jdbc_query_tmp):
                              jdbc_query_tmp1 = jdbc_query_tmp.replace("\n", "\\n")
                              jdbc_query = jdbc_query_tmp1.replace("\'", "\\'")
                              list.append(jdbc_query)
                          else:
                              jdbc_query = "0"
                              list.append(jdbc_query)
                      '''
                       
                       #获取query_id
                       if line["name"] == "hive.query.id":
                           #list.append(str(line["value"]))
                           query_id_tmp = line["value"]

                           if query_id_tmp:
                               query_id = line["value"]
                               list.append(query_id)
                           else:
                               query_id = "0"
                               list.append(query_id)
                               
                        #获取执行作业的队列名称,就能对应到租户
                       elif line["name"] == "mapreduce.job.queuename":
                           #list.append(str(line["value"]))
                           user_info = (line["source"])[1]
                           user = str(user_info).split("/")
                           queue_name = user[4]
                           list.append(queue_name)
                       else:
                           pass
                   print(list)
                   print(len(list))
                   try:
                       pass
                       ConnMysqlTmpl.user_app_query_info1(list)
                   except Exception as e:
                       print(f"数据入库失败,请排查:{e}")
               except JSONDecodeError as e:
                   print("{0}作业非MapReduce任务,请排查~".format(line))
           except Exception as e:
               print("获取app信息失败,请排查 {0}".format(e))
   else:
       print("获取app_id信息失败,请排查~")

主要说明

· 执行前(ACCEPTED)状态下的作业,内容编码存在异常,无法通过yarn rest api进行读取

· 执行完成的作业(FINISHED)的作业无异常情况下都会及时的释放锁,如果集群开始了日志服务,可以访问jobhistoryserver服务读取对应日志

获取运行状态下作业的信息,链接为:

/http://resouecemanage:port/applicationID/ws/v1/mapreduce/jobs/jobId/conf/

将返回的数据进行json格式转换:

info_json=all_info_tmp.json() result=info_json["conf"] all_info=result["property"]

获取query_id:

if line["name"] == "hive.query.id":
                           #list.append(str(line["value"]))
                           query_id_tmp = line["value"]

获取资源队列名称:

line["name"] == "mapreduce.job.queuename":
                           #list.append(str(line["value"]))
                           user_info = (line["source"])[1]
                           user = str(user_info).split("/")
                           queue_name = user[4]
                           list.append(queue_name)

效果呈现

SELECT
a.query_id,
a.queue_name,
b.sql_info,
b.table_name,
b.ctime
FROM
user_app_query_info a
JOIN zookeeper_lock_info b
WHERE
a.query_id = b.query_id

两个表根据query_id进行关联,获取到锁表情况下,锁表的SQL语句和租户信息。

标签:info,SQL,list,hive,sql,query,line,表锁,id
来源: https://www.cnblogs.com/zxyax132620/p/15514885.html