0
0

实际业务读取Hive数据库（2023年2月）

背景：

在这篇文章之前，我读取数据库的数据没有形成规范，并且代码扩展性不好，使用率不高，而且比较混乱。数据库信息的替换也比较混乱。坏习惯包括：连接数据库之后就开始读数，读完就结束，数据的存放也没有规范，而且容易重复读取。

现在将代码分为几层，一层是底层，就是单独连接数据库，在这基础上封装第二个类别，加上了线程锁和时间表，用于确保读数的稳定和超时错误提醒。第三层才是真正的业务，第三层的类里面封装了很多读取不同数据表的方法，每一个方法就是读一个表，然后将数据缓存起来，并且设置好更新数据缓存的时间（例如24小时），和维护多线程读数。

第四层也就是简单的调用第三层即可，然后所有的数据都可以读取然后缓存到我们在配置项中指定的文件夹目录了

文章目录

实际业务读取hive数据库的代码

  1. import logging
  2. import pandas as pd
  3. from impala.dbapi import connect
  4. import sqlalchemy
  5. from sqlalchemy.orm import sessionmaker
  6. import os
  7. import time
  9. import os
  10. import datetime
  11. from dateutil.relativedelta import relativedelta
  12. from typing import Dict, List
  13. import logging
  14. import threading
  15. import pandas as pd
  17. import pickle
  19. classHiveHelper(object):def__init__(
  20.         self,
  21.         host='10.2.32.22',
  22.         port=21051,
  23.         database='ur_ai_dw',
  24.         auth_mechanism='LDAP',
  25.         user='urbi',
  26.         password='Ur#730xd',
  27.         logger:logging.Logger=None):
  28.         self.host = host
  29.         self.port = port
  30.         self.database = database
  31.         self.auth_mechanism = auth_mechanism
  32.         self.user = user
  33.         self.password = password
  34.         self.logger = logger
  36.         self.impala_conn =None
  37.         self.conn =None
  38.         self.cursor =None
  39.         self.engine =None
  40.         self.session =Nonedefcreate_table_code(self, file_name):'''创建表类代码'''
  41.         os.system(f'sqlacodegen {self.connection_str} > {file_name}')return self.conn
  43.     defget_conn(self):'''创建连接或获取连接'''if self.conn isNone:
  44.             engine = self.get_engine()
  45.             self.conn = engine.connect()return self.conn
  47.     defget_impala_conn(self):'''创建连接或获取连接'''if self.impala_conn isNone:
  48.             self.impala_conn = connect(
  49.                 host=self.host,
  50.                 port=self.port,
  51.                 database=self.database,
  52.                 auth_mechanism=self.auth_mechanism,
  53.                 user=self.user,
  54.                 password=self.password
  55.                 )return self.impala_conn
  57.     defget_engine(self):'''创建连接或获取连接'''if self.engine isNone:
  58.             self.engine = sqlalchemy.create_engine('impala://', creator=self.get_impala_conn)return self.engine
  60.     defget_cursor(self):'''创建连接或获取连接'''if self.cursor isNone:
  61.             self.cursor = self.conn.cursor()return self.cursor
  63.     defget_session(self)-> sessionmaker:'''创建连接或获取连接'''if self.session isNone:
  64.             engine = self.get_engine()
  65.             Session = sessionmaker(bind=engine)
  66.             self.session = Session()return self.session
  67.         
  68.     defclose_conn(self):'''关闭连接'''if self.conn isnotNone:
  69.             self.conn.close()
  70.             self.conn =None
  71.         self.dispose_engine()
  72.         self.close_impala_conn()defclose_impala_conn(self):'''关闭impala连接'''if self.impala_conn isnotNone:
  73.             self.impala_conn.close()
  74.             self.impala_conn =Nonedefclose_session(self):'''关闭连接'''if self.session isnotNone:
  75.             self.session.close()
  76.             self.session =None
  77.         self.dispose_engine()defdispose_engine(self):'''释放engine'''if self.engine isnotNone:# self.engine.dispose(close=False)
  78.             self.engine.dispose()
  79.             self.engine =Nonedefclose_cursor(self):'''关闭cursor'''if self.cursor isnotNone:
  80.             self.cursor.close()
  81.             self.cursor =Nonedefget_data(self, sql, auto_close=True)-> pd.DataFrame:'''查询数据'''
  82.         conn = self.get_conn()
  83.         data =Nonetry:# 异常重试3次for i inrange(3):try:
  84.                     data = pd.read_sql(sql, conn)breakexcept Exception as ex:if i ==2:raise ex # 往外抛出异常
  85.                     time.sleep(60)# 一分钟后重试except Exception as ex:
  86.             self.logger.exception(ex)raise ex # 往外抛出异常finally:if auto_close:
  87.                 self.close_conn()return data
  88. passclassVarsHelper():def__init__(self, save_dir, auto_save=True):
  89.         self.save_dir = save_dir
  90.         self.auto_save = auto_save
  91.         self.values ={}ifnot os.path.exists(os.path.dirname(self.save_dir)):
  92.             os.makedirs(os.path.dirname(self.save_dir))if os.path.exists(self.save_dir):withopen(self.save_dir,'rb')as f:
  93.                 self.values = pickle.load(f)
  94.                 f.close()defset_value(self, key, value):
  95.         self.values[key]= value
  96.         if self.auto_save:
  97.             self.save_file()defget_value(self, key):return self.values[key]defhas_key(self, key):return key in self.values.keys()defsave_file(self):withopen(self.save_dir,'wb')as f:
  98.             pickle.dump(self.values, f)
  99.             f.close()passclassGlobalShareArgs():
  100.     args ={"debug":False}defget_args():return GlobalShareArgs.args
  102.     defset_args(args):
  103.         GlobalShareArgs.args = args
  105.     defset_args_value(key, value):
  106.         GlobalShareArgs.args[key]= value
  108.     defget_args_value(key, default_value=None):return GlobalShareArgs.args.get(key, default_value)defcontain_key(key):return key in GlobalShareArgs.args.keys()defupdate(args):
  109.         GlobalShareArgs.args.update(args)passclassShareArgs():
  110.     args ={"labels_dir":"./hjx/shop_group/month_w_amt/data/labels",# 标签目录"labels_output_dir":"./hjx/shop_group/month_w_amt/data/labels_output",# 聚类导出标签目录"common_datas_dir":"./hjx/data",# 共用数据目录。ur_bi_dw的公共"only_predict":False,# 只识别，不训练"delete_model":True,# 先删除模型，仅在训练时使用"export_excel":False,# 导出excel"classes":12,# 聚类数"batch_size":16,"hidden_size":32,"max_nrof_epochs":100,"learning_rate":0.0005,"loss_type":"categorical_crossentropy","avg_model_num":10,"steps_per_epoch":4.0,# 4.0"lr_callback_patience":4,"lr_callback_cooldown":1,"early_stopping_callback_patience":6,"get_data":True,}defget_args():return ShareArgs.args
  112.     defset_args(args):
  113.         ShareArgs.args = args
  115.     defset_args_value(key, value):
  116.         ShareArgs.args[key]= value
  118.     defget_args_value(key, default_value=None):return ShareArgs.args.get(key, default_value)defcontain_key(key):return key in ShareArgs.args.keys()defupdate(args):
  119.         ShareArgs.args.update(args)passclassUrBiGetDatasBase():# 线程锁列表，同保存路径共用锁
  120.     lock_dict:Dict[str, threading.Lock]={}# 时间列表，用于判断是否超时
  121.     time_dict:Dict[str, datetime.datetime]={}# 用于记录是否需要更新超时时间
  122.     get_data_timeout_dict:Dict[str,bool]={}def__init__(
  123.         self,
  124.         host='10.2.32.22',
  125.         port=21051,
  126.         database='ur_ai_dw',
  127.         auth_mechanism='LDAP',
  128.         user='urbi',
  129.         password='Ur#730xd',
  130.         save_dir=None,
  131.         logger:logging.Logger=None,):
  132.         self.save_dir = save_dir
  133.         self.logger = logger
  134.         self.db_helper = HiveHelper(
  135.             host=host,
  136.             port=port,
  137.             database=database,
  138.             auth_mechanism=auth_mechanism,
  139.             user=user,
  140.             password=password,
  141.             logger=logger
  142.             )# 创建子目录if self.save_dir isnotNoneandnot os.path.exists(self.save_dir):
  143.             os.makedirs(self.save_dir)
  144.         
  145.         self.vars_helper =Noneif GlobalShareArgs.get_args_value('debug'):
  146.             self.vars_helper = VarsHelper('./hjx/data/vars/UrBiGetDatas')defclose(self):'''关闭连接'''
  147.         self.db_helper.close_conn()defget_last_time(self, key_name)->bool:'''获取是否超时'''# 转静态路径，确保唯一性
  148.         key_name = os.path.abspath(key_name)if self.vars_helper isnotNoneand self.vars_helper.has_key('UrBiGetDatasBase.time_list'):
  149.             UrBiGetDatasBase.time_dict = self.vars_helper.get_value('UrBiGetDatasBase.time_list')
  150.         timeout =12# 12小时if GlobalShareArgs.get_args_value('debug'):
  151.             timeout =24# 24小时
  152.         get_data_timeout =Falseif key_name notin UrBiGetDatasBase.time_dict.keys()or(datetime.datetime.today()- UrBiGetDatasBase.time_dict[key_name]).total_seconds()>(timeout*60*60):
  153.             self.logger.info('超时%d小时，重新查数据：%s', timeout, key_name)# UrBiGetDatasBase.time_list[key_name] = datetime.datetime.today()
  154.             get_data_timeout =Trueelse:
  155.             self.logger.info('未超时%d小时，跳过查数据：%s', timeout, key_name)# if self.vars_helper is not None :#     self.vars_helper.set_value('UrBiGetDatasBase.time_list', UrBiGetDatasBase.time_list)
  156.         UrBiGetDatasBase.get_data_timeout_dict[key_name]= get_data_timeout
  157.         return get_data_timeout
  158.     
  159.     defsave_last_time(self, key_name):'''更新状态超时'''# 转静态路径，确保唯一性
  160.         key_name = os.path.abspath(key_name)if UrBiGetDatasBase.get_data_timeout_dict[key_name]:
  161.             UrBiGetDatasBase.time_dict[key_name]= datetime.datetime.today()if self.vars_helper isnotNone:
  162.             UrBiGetDatasBase.time_dict[key_name]= datetime.datetime.today()
  163.             self.vars_helper.set_value('UrBiGetDatasBase.time_list', UrBiGetDatasBase.time_dict)defget_lock(self, key_name)-> threading.Lock:'''获取锁'''# 转静态路径，确保唯一性
  164.         key_name = os.path.abspath(key_name)if key_name notin UrBiGetDatasBase.lock_dict.keys():
  165.             UrBiGetDatasBase.lock_dict[key_name]= threading.Lock()return UrBiGetDatasBase.lock_dict[key_name]defget_data_of_date(
  166.         self,
  167.         save_dir,
  168.         sql,
  169.         sort_columns:List[str],
  170.         del_index_list=[-1],# 删除最后下标
  171.         start_date = datetime.datetime(2017,1,1),# 开始时间
  172.         offset = relativedelta(months=3),# 时间间隔
  173.         date_format_fun =lambda d:'%04d%02d01'%(d.year, d.month),# 查询语句中替代时间参数的格式化
  174.         filename_format_fun =lambda d:'%04d%02d.csv'%(d.year, d.month),# 查询语句中替代时间参数的格式化
  175.         stop_date ='20700101',# 超过时间则停止
  176.         data_format_fun =None,# 格式化数据):'''分时间增量读取数据'''# 创建文件夹ifnot os.path.exists(save_dir):
  177.             os.makedirs(save_dir)else:#删除最后一个文件
  178.             file_list = os.listdir(save_dir)iflen(file_list)>0:
  179.                 file_list.sort()for del_index in del_index_list:
  180.                     os.remove(os.path.join(save_dir,file_list[del_index]))print('删除最后一个文件：', file_list[del_index])
  181.         select_index =-1# start_date = datetime.datetime(2017, 1, 1)whileTrue:
  182.             end_date = start_date + offset
  183.             start_date_str = date_format_fun(start_date)
  184.             end_date_str = date_format_fun(end_date)
  185.             self.logger.info('date: %s-%s', start_date_str, end_date_str)
  186.             file_path = os.path.join(save_dir, filename_format_fun(start_date))# self.logger.info('file_path: %s', file_path)ifnot os.path.exists(file_path):
  187.                 data:pd.DataFrame = self.db_helper.get_data(sql %(start_date_str, end_date_str))if data isNone:break
  188.                 self.logger.info('data: %d',len(data))# self.logger.info('data: %d', data.columns)iflen(data)>0:
  189.                     select_index+=1if data_format_fun isnotNone:
  190.                         data = data_format_fun(data)# 排序
  191.                     data = data.sort_values(sort_columns)
  192.                     data.to_csv(file_path)elif select_index!=-1:breakelif stop_date < start_date_str:raise Exception("读取数据异常，时间超出最大值！")
  193.             start_date = end_date
  194. passclassUrBiGetDatas(UrBiGetDatasBase):def__init__(
  195.         self,
  196.         host='10.2.32.22',
  197.         port=21051,
  198.         database='ur_ai_dw',
  199.         auth_mechanism='LDAP',
  200.         user='urbi',
  201.         password='Ur#730xd',
  202.         save_dir='./hjx/data/ur_bi_dw_data',
  203.         logger:logging.Logger=None):
  204.         self.save_dir = save_dir
  205.         self.logger = logger
  206.         super().__init__(
  207.             host=host,
  208.             port=port,
  209.             database=database,
  210.             auth_mechanism=auth_mechanism,
  211.             user=user,
  212.             password=password,
  213.             save_dir=save_dir,
  214.             logger=logger
  215.             )defget_dim_date(self):'''日期数据'''
  216.         file_path = os.path.join(self.save_dir,'ur_bi_dw.dim_date.csv')
  217.         now_lock = self.get_lock(file_path)
  218.         now_lock.acquire()# 加锁try:# 设置超时4小时才重新查数据ifnot self.get_last_time(file_path):return
  219.         
  220.             sql ='SELECT * FROM ur_bi_dw.dim_date'
  221.             data:pd.DataFrame = self.db_helper.get_data(sql)
  222.             columns =list(data.columns)
  223.             columns ={c:'dim_date.'+c for c in columns}
  224.             data = data.rename(columns=columns)
  225.             data = data.sort_values(['dim_date.date_key'])
  226.             data.to_csv(file_path)# 更新超时时间
  227.             self.save_last_time(file_path)except Exception as ex:
  228.             self.logger.exception(ex)raise ex # 往外抛出异常finally:
  229.             now_lock.release()# 释放锁defget_dim_shop(self):'''店铺数据'''
  230.         file_path = os.path.join(self.save_dir,'ur_bi_dw.dim_shop.csv')
  231.         now_lock = self.get_lock(file_path)
  232.         now_lock.acquire()# 加锁try:# 设置超时4小时才重新查数据ifnot self.get_last_time(file_path):return
  233.         
  234.             sql ='SELECT * FROM ur_bi_dw.dim_shop'
  235.             data:pd.DataFrame = self.db_helper.get_data(sql)
  236.             columns =list(data.columns)
  237.             columns ={c:'dim_shop.'+c for c in columns}
  238.             data = data.rename(columns=columns)
  239.             data = data.sort_values(['dim_shop.shop_no'])
  240.             data.to_csv(file_path)# 更新超时时间
  241.             self.save_last_time(file_path)except Exception as ex:
  242.             self.logger.exception(ex)raise ex # 往外抛出异常finally:
  243.             now_lock.release()# 释放锁defget_dim_vip(self):'''会员数据'''
  244.         sub_dir = os.path.join(self.save_dir,'vip_no')
  245.         now_lock = self.get_lock(sub_dir)
  246.         now_lock.acquire()# 加锁try:# 设置超时4小时才重新查数据ifnot self.get_last_time(sub_dir):return
  247.             sql ='''SELECT dv.*, dd.date_key, dd.date_name2 
  248.             FROM ur_bi_dw.dim_vip as dv
  249.             INNER JOIN ur_bi_dw.dim_date as dd
  250.             ON dv.card_create_date=dd.date_name2 
  251.             where dd.date_key >= %s
  252.             and dd.date_key < %s'''# data:pd.DataFrame = self.db_helper.get_data(sql)
  253.             sort_columns =['dv.vip_no']# TODO:
  254.             self.get_data_of_date(
  255.                 save_dir=sub_dir,
  256.                 sql=sql,
  257.                 sort_columns=sort_columns,
  258.                 start_date=datetime.datetime(2017,1,1),# 开始时间
  259.                 offset=relativedelta(years=1))# 更新超时时间
  260.             self.save_last_time(sub_dir)except Exception as ex:
  261.             self.logger.exception(ex)raise ex # 往外抛出异常finally:
  262.             now_lock.release()# 释放锁defget_weather(self):'''天气数据'''
  263.         sub_dir = os.path.join(self.save_dir,'weather')
  264.         now_lock = self.get_lock(sub_dir)
  265.         now_lock.acquire()# 加锁try:# 设置超时4小时才重新查数据ifnot self.get_last_time(sub_dir):return
  266.         
  267.             sql ="""
  268.             select weather.* from ur_bi_ods.ods_base_weather_data_1200 as weather
  269.             where weather.date_key>=%s and weather.date_key<%s
  270.             """
  271.             sort_columns =['weather.date_key','weather.areaid']defdata_format_fun(data):
  272.                 columns =list(data.columns)
  273.                 columns ={c:'weather.'+c for c in columns}
  274.                 data = data.rename(columns=columns)return data
  276.             self.get_data_of_date(
  277.                 save_dir=sub_dir,
  278.                 sql=sql,
  279.                 sort_columns=sort_columns,
  280.                 del_index_list=[-2,-1],# 删除最后下标
  281.                 data_format_fun=data_format_fun,)# 更新超时时间
  282.             self.save_last_time(sub_dir)except Exception as ex:
  283.             self.logger.exception(ex)raise ex # 往外抛出异常finally:
  284.             now_lock.release()# 释放锁defget_weather_city(self):'''天气城市数据'''
  285.         file_path = os.path.join(self.save_dir,'ur_bi_dw.weather_city.csv')
  286.         now_lock = self.get_lock(file_path)
  287.         now_lock.acquire()# 加锁try:# 设置超时4小时才重新查数据ifnot self.get_last_time(file_path):return
  288.         
  289.             sql ='SELECT * FROM ur_bi_dw.dim_weather_city as weather_city'
  290.             data:pd.DataFrame = self.db_helper.get_data(sql)
  291.             columns =list(data.columns)
  292.             columns ={c:'weather_city.'+c for c in columns}
  293.             data = data.rename(columns=columns)
  294.             data.to_csv(file_path)# 更新超时时间
  295.             self.save_last_time(file_path)except Exception as ex:
  296.             self.logger.exception(ex)raise ex # 往外抛出异常finally:
  297.             now_lock.release()# 释放锁defget_dim_goods(self):'''货品数据'''
  298.         file_path = os.path.join(self.save_dir,'ur_bi_dw.dim_goods.csv')
  299.         now_lock = self.get_lock(file_path)
  300.         now_lock.acquire()# 加锁try:# 设置超时4小时才重新查数据ifnot self.get_last_time(file_path):return
  301.         
  302.             sql ='SELECT * FROM ur_bi_dw.dim_goods'
  303.             data:pd.DataFrame = self.db_helper.get_data(sql)
  304.             columns =list(data.columns)
  305.             columns ={c:'dim_goods.'+c for c in columns}
  306.             data = data.rename(columns=columns)
  307.             data.to_csv(file_path)# 更新超时时间
  308.             self.save_last_time(file_path)except Exception as ex:
  309.             self.logger.exception(ex)raise ex # 往外抛出异常finally:
  310.             now_lock.release()# 释放锁defget_dim_goods_market_shop_date(self):'''店铺商品生命周期数据'''
  311.         file_path = os.path.join(self.save_dir,'ur_bi_dw.dim_goods_market_shop_date.csv')
  312.         now_lock = self.get_lock(file_path)
  313.         now_lock.acquire()# 加锁try:# 设置超时4小时才重新查数据ifnot self.get_last_time(file_path):return# sql = 'SELECT * FROM ur_bi_dw.dim_goods_market_shop_date as goods_shop_date'
  314.             sql ='''
  315.             select shop_no, sku_no, shop_market_date, lifecycle_end_date, lifecycle_days
  316.             FROM ur_bi_dw.dim_goods_market_shop_date
  317.             where lifecycle_end_date is not null
  318.             '''
  319.             data:pd.DataFrame = self.db_helper.get_data(sql)
  320.             columns =list(data.columns)
  321.             columns ={c:c.replace('lifecycle_end_date.','')for c in columns}
  322.             data = data.rename(columns=columns)
  323.             data = data.sort_values(['shop_market_date'])
  324.             data.to_csv(file_path, index=False)# 更新超时时间
  325.             self.save_last_time(file_path)except Exception as ex:
  326.             self.logger.exception(ex)raise ex # 往外抛出异常finally:
  327.             now_lock.release()# 释放锁defget_dim_goods_market_date(self):'''全国商品生命周期数据'''
  328.         file_path = os.path.join(self.save_dir,'ur_bi_dw.dim_goods_market_date.csv')
  329.         now_lock = self.get_lock(file_path)
  330.         now_lock.acquire()# 加锁try:# 设置超时4小时才重新查数据ifnot self.get_last_time(file_path):return
  331.         
  332.             sql ='''
  333.             select * FROM ur_bi_dw.dim_goods_market_date
  334.             '''
  335.             data:pd.DataFrame = self.db_helper.get_data(sql)
  336.             columns =list(data.columns)
  337.             columns ={c:'dim_goods_market_date.'+c for c in columns}
  338.             data = data.rename(columns=columns)
  339.             data = data.sort_values(['dim_goods_market_date.sku_no'])
  340.             data.to_csv(file_path, index=False)# 更新超时时间
  341.             self.save_last_time(file_path)except Exception as ex:
  342.             self.logger.exception(ex)raise ex # 往外抛出异常finally:
  343.             now_lock.release()# 释放锁defget_dim_goods_color_dev_sizes(self):'''商品开发码数数据'''
  344.         file_path = os.path.join(self.save_dir,'dim_goods_color_dev_sizes.csv')
  345.         now_lock = self.get_lock(file_path)
  346.         now_lock.acquire()# 加锁try:# 设置超时4小时才重新查数据ifnot self.get_last_time(file_path):return# sql = 'SELECT * FROM ur_bi_dw.dim_goods_market_shop_date as goods_shop_date'
  347.             sql ='SELECT * FROM ur_bi_dm.dim_goods_color_dev_sizes'
  348.             data:pd.DataFrame = self.db_helper.get_data(sql)
  349.             columns =list(data.columns)
  350.             columns ={c:c.replace('dim_goods_color_dev_sizes.','')for c in columns}
  351.             data = data.rename(columns=columns)
  352.             data.to_csv(file_path, index=False)# 更新超时时间
  353.             self.save_last_time(file_path)except Exception as ex:
  354.             self.logger.exception(ex)raise ex # 往外抛出异常finally:
  355.             now_lock.release()# 释放锁defget_dwd_daily_sales_size(self):'''实际销售金额'''
  356.         sub_dir = os.path.join(self.save_dir,'dwd_daily_sales_size_all')
  357.         now_lock = self.get_lock(sub_dir)
  358.         now_lock.acquire()# 加锁try:# 设置超时4小时才重新查数据ifnot self.get_last_time(sub_dir):return
  359.             sql ="""
  360.             select shop_no,sku_no,date_key,`size`,
  361.                 sum(tag_price) as `tag_price`,
  362.                 sum(sales_qty) as `sales_qty`,
  363.                 sum(sales_tag_amt) as `sales_tag_amt`,
  364.                 sum(sales_amt) as `sales_amt`,
  365.                 count(0) as `sales_count`
  366.             from ur_bi_dw.dwd_daily_sales_size as sales
  367.             where sales.date_key>=%s and sales.date_key<%s
  368.                 and sales.currency_code='CNY'
  369.             group by shop_no,sku_no,date_key,`size`
  370.             """
  371.             sort_columns =['date_key','shop_no','sku_no']
  372.             self.get_data_of_date(
  373.                 save_dir=sub_dir,
  374.                 sql=sql,
  375.                 sort_columns=sort_columns,
  376.                 start_date=datetime.datetime(2017,1,1),# 开始时间)# 更新超时时间
  377.             self.save_last_time(sub_dir)except Exception as ex:
  378.             self.logger.exception(ex)raise ex # 往外抛出异常finally:
  379.             now_lock.release()# 释放锁defget_dwd_daily_delivery_size(self):'''实际配货金额'''
  380.         sub_dir = os.path.join(self.save_dir,'dwd_daily_delivery_size_all')
  381.         now_lock = self.get_lock(sub_dir)
  382.         now_lock.acquire()# 加锁try:# 设置超时4小时才重新查数据ifnot self.get_last_time(sub_dir):return
  383.             sql ="""
  384.             select shop_no,sku_no,date_key,`size`,
  385.                 sum(delivery.shop_distr_received_qty) as `shop_distr_received_qty`,
  386.                 sum(delivery.shop_distr_received_amt) as `shop_distr_received_amt`,
  387.                 sum(delivery.online_distr_received_qty) as `online_distr_received_qty`,
  388.                 sum(delivery.online_distr_received_amt) as `online_distr_received_amt`,
  389.                 sum(delivery.pr_received_qty) as `pr_received_qty`,
  390.                 count(0) as `delivery_count`
  391.             from ur_bi_dw.dwd_daily_delivery_size as delivery
  392.             where delivery.date_key>=%s and delivery.date_key<%s
  393.                 and delivery.currency_code='CNY'
  394.             group by shop_no,sku_no,date_key,`size`
  395.             """
  396.             sort_columns =['date_key','shop_no','sku_no']
  397.             self.get_data_of_date(
  398.                 save_dir=sub_dir,
  399.                 sql=sql,
  400.                 sort_columns=sort_columns,
  401.                 start_date=datetime.datetime(2017,1,1),# 开始时间)# 更新超时时间
  402.             self.save_last_time(sub_dir)except Exception as ex:
  403.             self.logger.exception(ex)raise ex # 往外抛出异常finally:
  404.             now_lock.release()# 释放锁defget_v_last_nation_sales_status(self):'''商品畅滞销数据'''
  405.         file_path = os.path.join(self.save_dir,'v_last_nation_sales_status.csv')
  406.         now_lock = self.get_lock(file_path)
  407.         now_lock.acquire()# 加锁try:# 设置超时4小时才重新查数据ifnot self.get_last_time(file_path):return
  408.         
  409.             sql ='SELECT * FROM ur_bi_dw.v_last_nation_sales_status'
  410.             data:pd.DataFrame = self.db_helper.get_data(sql)
  411.             columns =list(data.columns)
  412.             columns ={c:c.replace('v_last_nation_sales_status.','')for c in columns}
  413.             data = data.rename(columns=columns)
  414.             data.to_csv(file_path, index=False)# 更新超时时间
  415.             self.save_last_time(file_path)except Exception as ex:
  416.             self.logger.exception(ex)raise ex # 往外抛出异常finally:
  417.             now_lock.release()# 释放锁defget_dwd_daily_finacial_goods(self):'''商品成本价数据'''
  418.         file_path = os.path.join(self.save_dir,'dwd_daily_finacial_goods.csv')
  419.         now_lock = self.get_lock(file_path)
  420.         now_lock.acquire()# 加锁try:# 设置超时4小时才重新查数据ifnot self.get_last_time(file_path):return
  421.         
  422.             sql ="""
  423.             select t1.sku_no,t1.`size`,t1.cost_tax_incl from ur_bi_dw.dwd_daily_finacial_goods as t1
  424.             inner join (
  425.                 select sku_no,`size`,max(date_key) as date_key
  426.                 from ur_bi_dw.dwd_daily_finacial_goods
  427.                 where currency_code='CNY' and country_code='CN'
  428.                 group by sku_no,`size`
  429.             ) as t2
  430.             on t2.sku_no=t1.sku_no
  431.                 and t2.`size`=t1.`size`
  432.                 and t2.date_key=t1.date_key
  433.             where t1.currency_code='CNY' and t1.country_code='CN'
  434.             """
  435.             data:pd.DataFrame = self.db_helper.get_data(sql)
  436.             columns =list(data.columns)
  437.             columns ={c:c.replace('t1.','')for c in columns}
  438.             data = data.rename(columns=columns)
  439.             data.to_csv(file_path, index=False)# 更新超时时间
  440.             self.save_last_time(file_path)except Exception as ex:
  441.             self.logger.exception(ex)raise ex # 往外抛出异常finally:
  442.             now_lock.release()# 释放锁defget_dim_size_group(self):'''尺码映射数据'''
  443.         file_path = os.path.join(self.save_dir,'dim_size_group.csv')
  444.         now_lock = self.get_lock(file_path)
  445.         now_lock.acquire()# 加锁try:# 设置超时4小时才重新查数据ifnot self.get_last_time(file_path):return
  446.         
  447.             sql ="""select * from ur_bi_dw.dim_size_group"""
  448.             data:pd.DataFrame = self.db_helper.get_data(sql)
  449.             columns =list(data.columns)
  450.             columns ={c:c.replace('dim_size_group.','')for c in columns}
  451.             data = data.rename(columns=columns)
  452.             data.to_csv(file_path, index=False)# 更新超时时间
  453.             self.save_last_time(file_path)except Exception as ex:
  454.             self.logger.exception(ex)raise ex # 往外抛出异常finally:
  455.             now_lock.release()# 释放锁passdefget_common_datas(
  456.     host='10.2.32.22',
  457.     port=21051,
  458.     database='ur_ai_dw',
  459.     auth_mechanism='LDAP',
  460.     user='urbi',
  461.     password='Ur#730xd',
  462.     logger:logging.Logger=None):# 共用文件
  463.     common_datas_dir = ShareArgs.get_args_value('common_datas_dir')
  464.     common_ur_bi_dir = os.path.join(common_datas_dir,'ur_bi_data')
  465.     ur_bi_get_datas = UrBiGetDatas(
  466.         host=host,
  467.         port=port,
  468.         database=database,
  469.         auth_mechanism=auth_mechanism,
  470.         user=user,
  471.         password=password,
  472.         save_dir=common_ur_bi_dir,
  473.         logger=logger
  474.     )try:
  475.         logger.info('正在查询日期数据...')
  476.         ur_bi_get_datas.get_dim_date()
  477.         logger.info('查询日期数据完成！')
  479.         logger.info('正在查询店铺数据...')
  480.         ur_bi_get_datas.get_dim_shop()
  481.         logger.info('查询店铺数据完成！')
  483.         logger.info('正在查询天气数据...')
  484.         ur_bi_get_datas.get_weather()
  485.         logger.info('查询天气数据完成！')
  487.         logger.info('正在查询天气城市数据...')
  488.         ur_bi_get_datas.get_weather_city()
  489.         logger.info('查询天气城市数据完成！')
  491.         logger.info('正在查询货品数据...')
  492.         ur_bi_get_datas.get_dim_goods()
  493.         logger.info('查询货品数据完成！')
  494.         
  495.         logger.info('正在查询实际销量数据...')
  496.         ur_bi_get_datas.get_dwd_daily_sales_size()
  497.         logger.info('查询实际销量数据完成！')except Exception as ex:
  498.         logger.exception(ex)raise ex # 往外抛出异常finally:
  499.         ur_bi_get_datas.close()passclassCustomUrBiGetDatas(UrBiGetDatasBase):def__init__(
  500.         self,
  501.         host='10.2.32.22',
  502.         port=21051,
  503.         database='ur_ai_dw',
  504.         auth_mechanism='LDAP',
  505.         user='urbi',
  506.         password='Ur#730xd',
  507.         save_dir='./hjx/data/ur_bi_data',
  508.         logger:logging.Logger=None):
  509.         self.save_dir = save_dir
  510.         self.logger = logger
  511.         super().__init__(
  512.             host=host,
  513.             port=port,
  514.             database=database,
  515.             auth_mechanism=auth_mechanism,
  516.             user=user,
  517.             password=password,
  518.             save_dir=save_dir,
  519.             logger=logger
  520.             )defget_sales_goal_amt(self):'''销售目标金额'''
  521.         file_path = os.path.join(self.save_dir,'month_of_year_sales_goal_amt.csv')
  522.         now_lock = self.get_lock(file_path)
  523.         now_lock.acquire()# 加锁try:# 设置超时4小时才重新查数据ifnot self.get_last_time(file_path):return
  524.         
  525.             sql ='''
  526.             select sales_goal.shop_no,
  527.                 if(sales_goal.serial='Y','W',sales_goal.serial) as `sales_goal.serial`,
  528.                 dates.month_of_year,
  529.                 sum(sales_goal.sales_goal_amt) as sales_goal_amt
  530.             from ur_bi_dw.dwd_sales_goal_west as sales_goal
  531.             inner join ur_bi_dw.dim_date as dates
  532.                 on sales_goal.date_key = dates.date_key
  533.             group by sales_goal.shop_no,
  534.                 if(sales_goal.serial='Y','W',sales_goal.serial),
  535.                 dates.month_of_year
  536.             '''
  537.             data:pd.DataFrame = self.db_helper.get_data(sql)
  538.             data = data.rename(columns={'shop_no':'sales_goal.shop_no','serial':'sales_goal.serial','month_of_year':'dates.month_of_year',})# 排序
  539.             data = data.sort_values(['sales_goal.shop_no','sales_goal.serial','dates.month_of_year'])
  540.             data.to_csv(file_path)# 更新超时时间
  541.             self.save_last_time(file_path)except Exception as ex:
  542.             self.logger.exception(ex)raise ex # 往外抛出异常finally:
  543.             now_lock.release()# 释放锁defget_shop_serial_area(self):'''店-系列面积'''
  544.         file_path = os.path.join(self.save_dir,'shop_serial_area.csv')
  545.         now_lock = self.get_lock(file_path)
  546.         now_lock.acquire()# 加锁try:# 设置超时4小时才重新查数据ifnot self.get_last_time(file_path):return
  547.         
  548.             sql ='''
  549.             select shop_serial_area.shop_no,
  550.                 if(shop_serial_area.serial='Y','W',shop_serial_area.serial) as `shop_serial_area.serial`,
  551.                 shop_serial_area.month_of_year,
  552.                 sum(shop_serial_area.area) as `shop_serial_area.area`
  553.             from ur_bi_dw.dwd_shop_serial_area as shop_serial_area
  554.             where shop_serial_area.area is not null
  555.             group by shop_serial_area.shop_no,if(shop_serial_area.serial='Y','W',shop_serial_area.serial),shop_serial_area.month_of_year
  556.             '''
  557.             data:pd.DataFrame = self.db_helper.get_data(sql)
  558.             data = data.rename(columns={'shop_no':'shop_serial_area.shop_no','serial':'shop_serial_area.serial','month_of_year':'shop_serial_area.month_of_year','area':'shop_serial_area.area',})# 排序
  559.             data = data.sort_values(['shop_serial_area.shop_no','shop_serial_area.serial','shop_serial_area.month_of_year'])
  560.             data.to_csv(file_path)# 更新超时时间
  561.             self.save_last_time(file_path)except Exception as ex:
  562.             self.logger.exception(ex)raise ex # 往外抛出异常finally:
  563.             now_lock.release()# 释放锁passdefget_datas(
  564.     host='10.2.32.22',
  565.     port=21051,
  566.     database='ur_ai_dw',
  567.     auth_mechanism='LDAP',
  568.     user='urbi',
  569.     password='Ur#730xd',
  570.     save_dir='./data/sales_forecast/ur_bi_dw_data',
  571.     logger:logging.Logger=None):
  572.     ur_bi_get_datas = CustomUrBiGetDatas(
  573.         host=host,
  574.         port=port,
  575.         database=database,
  576.         auth_mechanism=auth_mechanism,
  577.         user=user,
  578.         password=password,
  579.         save_dir=save_dir,
  580.         logger=logger
  581.     )try:# 店，系列，品类，年月，销售目标金额
  582.         logger.info('正在查询年月销售目标金额数据...')
  583.         ur_bi_get_datas.get_sales_goal_amt()
  584.         logger.info('查询年月销售目标金额数据完成！')except Exception as ex:
  585.         logger.exception(ex)raise ex # 往外抛出异常finally:
  586.         ur_bi_get_datas.close()passdefgetdata_ur_bi_dw(
  587.     host='10.2.32.22',
  588.     port=21051,
  589.     database='ur_ai_dw',
  590.     auth_mechanism='LDAP',
  591.     user='urbi',
  592.     password='Ur#730xd',
  593.     save_dir='./data/sales_forecast/ur_bi_dw_data',
  594.     logger=None):
  595.     get_common_datas(
  596.         host=host,
  597.         port=port,
  598.         database=database,
  599.         auth_mechanism=auth_mechanism,
  600.         user=user,
  601.         password=password,
  602.         logger=logger
  603.     )
  604.     get_datas(
  605.         host=host,
  606.         port=port,
  607.         database=database,
  608.         auth_mechanism=auth_mechanism,
  609.         user=user,
  610.         password=password,
  611.         save_dir=save_dir,
  612.         logger=logger
  613.     )pass# 代码入口# getdata_ur_bi_dw(#     host=ur_bi_dw_host,#     port=ur_bi_dw_port,#     database=ur_bi_dw_database,#     auth_mechanism=ur_bi_dw_auth_mechanism,#     user=ur_bi_dw_user,#     password=ur_bi_dw_password,#     save_dir=ur_bi_dw_save_dir,#     logger=logger#     )

代码说明和领悟

每个类的具体作用说明，代码需要根据下面的文字说明进行“食用”：

（第一层）HiveHelper完成了连接数据库、关闭数据库连接、生成事务、执行、引擎、连接等功能

VarsHelper提供了一个简单的持久化功能，可以将对象以文件的形式存放在磁盘上。并提供设置值、获取值、判断值是否存在的方法

GlobalShareArgs提供了一个字典，并且提供了获取字典、设置字典、设置字典键值对、设置字典键的值、判断键是否在字典中、更新字典等方法

ShareArgs跟GlobalShareArgs类似，只是一开始字典的初始化的键值对比较多

（第二层）UrBiGetDataBase类，提供了线程锁字典、时间字典、超时判断字典，都是类变量；使用了HiveHelper类，但注意，不是继承。在具体的sql读数时，提供了线程固定和时间判断

（第三层）UrBiGetDatas类，获取hive数据库那边的日期数据、店铺数据、会员数据、天气数据、天气城市数据、商品数据、店铺生命周期数据、全国商品生命周期数据、商品开发码数数据、实际销售金额、实际配货金额、商品畅滞销数据、商品成本价数据、尺码映射数据等。

（第四层）get_common_data函数，使用URBiGetData类读取日期、店铺、天气、天气城市、货品、实际销量数据，并缓存到文件夹./yongjian/data/ur_bi_data下面

CustomUrBiGetData类，继承了UrBiGetDatasBase类，读取销售目标金额、点系列面积数据。

（这个也是第四层）get_datas函数，通过CustomUrBiGetData类，读取年月销售目标金额。

总的函数：（这个是总的调用入口函数）get_data_ur_bi_dw函数，调用了get_common_data和get_datas函数进行读取数据，然后将数据保存到某个文件夹目录下面。

举一反三，如果你不是hive数据库，你可以将第一层这个底层更换成mysql。主页有解释如果进行更换。第二层不需要改变，第三层就是你想要进行读取的数据表，不同的数据库你想要读取的数据表也不同，所以sql需要你在这里写，套用里面的方法即可，基本上就是修改sql就好了。

这种方法的好处在于，数据不会重复读取，并且读取的数据都可以得到高效的使用。

后续附上修改成mysql的一个例子代码

  2. import logging
  3. import pandas as pd
  4. from impala.dbapi import connect
  5. import sqlalchemy
  6. from sqlalchemy.orm import sessionmaker
  7. import os
  8. import time
  10. import os
  11. import datetime
  12. from dateutil.relativedelta import relativedelta
  13. from typing import Dict, List
  14. import logging
  15. import threading
  16. import pandas as pd
  18. import pickle
  19. class MySqlHelper(object):
  20.     def __init__(
  21.         self,
  22.         host='192.168.15.144',
  23.         port=3306,
  24.         database='test_ims',
  25.         user='spkjz_writer',
  26.         password='7cmoP3QDtueVJQj2q4Az',
  27.         logger:logging.Logger=None
  28.         ):
  29.         self.host = host
  30.         self.port = port
  31.         self.database = database
  32.         self.user = user
  33.         self.password = password
  34.         self.logger = logger
  36.         self.connection_str = 'mysql+pymysql://%s:%s@%s:%d/%s' %(
  37.             self.user, self.password, self.host, self.port, self.database
  38.         )
  40.         self.conn = None
  41.         self.cursor = None
  42.         self.engine = None
  43.         self.session = None
  45.     def create_table_code(self, file_name):
  46.         '''创建表类代码'''
  47.         os.system(f'sqlacodegen {self.connection_str} > {file_name}')
  48.         return self.conn
  50.     def get_conn(self):
  51.         '''创建连接或获取连接'''
  52.         if self.conn is None:
  53.             engine = self.get_engine()
  54.             self.conn = engine.connect()
  55.         return self.conn
  57.     def get_engine(self):
  58.         '''创建连接或获取连接'''
  59.         if self.engine is None:
  60.             self.engine = sqlalchemy.create_engine(self.connection_str)
  61.         return self.engine
  63.     def get_cursor(self):
  64.         '''创建连接或获取连接'''
  65.         if self.cursor is None:
  66.             self.cursor = self.conn.cursor()
  67.         return self.cursor
  69.     def get_session(self) -> sessionmaker:
  70.         '''创建连接或获取连接'''
  71.         if self.session is None:
  72.             engine = self.get_engine()
  73.             Session = sessionmaker(bind=engine)
  74.             self.session = Session()
  75.         return self.session
  76.         
  77.     def close_conn(self):
  78.         '''关闭连接'''
  79.         if self.conn is not None:
  80.             self.conn.close()
  81.             self.conn = None
  82.         self.dispose_engine()
  83.         
  84.     def close_session(self):
  85.         '''关闭连接'''
  86.         if self.session is not None:
  87.             self.session.close()
  88.             self.session = None
  89.         self.dispose_engine()
  90.         
  91.     def dispose_engine(self):
  92.         '''释放engine'''
  93.         if self.engine is not None:
  94.             # self.engine.dispose(close=False)
  95.             self.engine.dispose()
  96.             self.engine = None
  97.         
  98.     def close_cursor(self):
  99.         '''关闭cursor'''
  100.         if self.cursor is not None:
  101.             self.cursor.close()
  102.             self.cursor = None
  104.     def get_data(self, sql, auto_close=True) -> pd.DataFrame:
  105.         '''查询数据'''
  106.         conn = self.get_conn()
  107.         data = None
  108.         try:
  109.             # 异常重试3次
  110.             for i in range(3):
  111.                 try:
  112.                     data = pd.read_sql(sql, conn)
  113.                     break
  114.                 except Exception as ex:
  115.                     if i == 2:
  116.                         raise ex # 往外抛出异常
  117.                     time.sleep(60) # 一分钟后重试
  118.         except Exception as ex:
  119.             self.logger.exception(ex)
  120.             raise ex # 往外抛出异常
  121.         finally:
  122.             if auto_close:
  123.                 self.close_conn()
  124.         return data
  125. pass
  126. class VarsHelper():
  127.     def __init__(self, save_dir, auto_save=True):
  128.         self.save_dir = save_dir
  129.         self.auto_save = auto_save
  130.         self.values = {}
  131.         if not os.path.exists(os.path.dirname(self.save_dir)):
  132.             os.makedirs(os.path.dirname(self.save_dir))
  133.         if os.path.exists(self.save_dir):
  134.             with open(self.save_dir, 'rb') as f:
  135.                 self.values = pickle.load(f)
  136.                 f.close()
  138.     def set_value(self, key, value):
  139.         self.values[key] = value
  140.         if self.auto_save:
  141.             self.save_file()
  143.     def get_value(self, key):
  144.         return self.values[key]
  146.     def has_key(self, key):
  147.         return key in self.values.keys()
  148.     
  149.     def save_file(self):
  150.         with open(self.save_dir, 'wb') as f:
  151.             pickle.dump(self.values, f)
  152.             f.close()
  153. pass
  154. class GlobalShareArgs():
  155.     args = {
  156.         "debug": False
  157.     }
  158.     def get_args():
  159.         return GlobalShareArgs.args
  161.     def set_args(args):
  162.         GlobalShareArgs.args = args
  164.     def set_args_value(key, value):
  165.         GlobalShareArgs.args[key] = value
  167.     def get_args_value(key, default_value=None):
  168.         return GlobalShareArgs.args.get(key, default_value)
  170.     def contain_key(key):
  171.         return key in GlobalShareArgs.args.keys()
  173.     def update(args):
  174.         GlobalShareArgs.args.update(args)
  175. pass
  176. class ShareArgs():
  177.     args = {
  178.         "labels_dir":"./hjx/shop_group/month_w_amt/data/labels", # 标签目录
  179.         "labels_output_dir":"./hjx/shop_group/month_w_amt/data/labels_output", # 聚类导出标签目录
  180.         "common_datas_dir":"./hjx/data", # 共用数据目录。ur_bi_dw的公共
  181.         "only_predict": False, # 只识别，不训练
  182.         "delete_model": True, # 先删除模型，仅在训练时使用
  183.         "export_excel": False, # 导出excel
  184.         "classes": 12, # 聚类数
  185.         "batch_size": 16,
  186.         "hidden_size": 32,
  187.         "max_nrof_epochs": 100,
  188.         "learning_rate": 0.0005,
  189.         "loss_type": "categorical_crossentropy",
  190.         "avg_model_num": 10,
  191.         "steps_per_epoch": 4.0, # 4.0
  192.         "lr_callback_patience": 4, 
  193.         "lr_callback_cooldown": 1,
  194.         "early_stopping_callback_patience": 6,
  195.         "get_data": True,
  196.     }
  197.     def get_args():
  198.         return ShareArgs.args
  200.     def set_args(args):
  201.         ShareArgs.args = args
  203.     def set_args_value(key, value):
  204.         ShareArgs.args[key] = value
  206.     def get_args_value(key, default_value=None):
  207.         return ShareArgs.args.get(key, default_value)
  209.     def contain_key(key):
  210.         return key in ShareArgs.args.keys()
  212.     def update(args):
  213.         ShareArgs.args.update(args)
  214. pass
  215. class IMSGetDatasBase():
  216.     # 线程锁列表，同保存路径共用锁
  217.     lock_dict:Dict[str, threading.Lock] = {}
  218.     # 时间列表，用于判断是否超时
  219.     time_dict:Dict[str, datetime.datetime] = {}
  220.     # 用于记录是否需要更新超时时间
  221.     get_data_timeout_dict:Dict[str, bool] = {}
  223.     def __init__(
  224.         self,
  225.         host='192.168.15.144',
  226.         port=3306,
  227.         database='test_ims',
  228.         user='spkjz_writer',
  229.         password='Ur#7cmoP3QDtueVJQj2q4Az',
  230.         save_dir=None,
  231.         logger:logging.Logger=None,
  232.         ):
  233.         self.save_dir = save_dir
  234.         self.logger = logger
  235.         self.db_helper = MySqlHelper(
  236.             host=host,
  237.             port=port,
  238.             database=database,
  239.             user=user,
  240.             password=password,
  241.             logger=logger
  242.             )
  243.         # 创建子目录
  244.         if self.save_dir is not None and not os.path.exists(self.save_dir):
  245.             os.makedirs(self.save_dir)
  246.     
  247.         self.vars_helper = None
  248.         if GlobalShareArgs.get_args_value('debug'):
  249.             self.vars_helper = VarsHelper('./hjx/data/vars/IMSGetDatas') # 把超时时间保存到文件，注释该行即可停掉，只用于调试
  250.     
  251.     def close(self):
  252.         '''关闭连接'''
  253.         self.db_helper.close_conn()
  254.     
  255.     def get_last_time(self, key_name) -> bool:
  256.         '''获取是否超时'''
  257.         # 转静态路径，确保唯一性
  258.         key_name = os.path.abspath(key_name)
  259.         if self.vars_helper is not None and self.vars_helper.has_key('IMSGetDatasBase.time_list'):
  260.             IMSGetDatasBase.time_dict = self.vars_helper.get_value('IMSGetDatasBase.time_list')
  261.         timeout = 12 # 12小时
  262.         if GlobalShareArgs.get_args_value('debug'):
  263.             timeout = 24 # 24小时
  264.         get_data_timeout = False
  265.         if key_name not in IMSGetDatasBase.time_dict.keys() or (datetime.datetime.today() - IMSGetDatasBase.time_dict[key_name]).total_seconds()>(4*60*60):
  266.             self.logger.info('超时%d小时，重新查数据：%s', timeout, key_name)
  267.             # IMSGetDatasBase.time_list[key_name] = datetime.datetime.today()
  268.             get_data_timeout = True
  269.         else:
  270.             self.logger.info('未超时%d小时，跳过查数据：%s', timeout, key_name)
  271.         # if self.vars_helper is not None :
  272.         #     self.vars_helper.set_value('IMSGetDatasBase.time_list', IMSGetDatasBase.time_list)
  273.         IMSGetDatasBase.get_data_timeout_dict[key_name] = get_data_timeout
  274.         return get_data_timeout
  275.     
  276.     def save_last_time(self, key_name):
  277.         '''更新状态超时'''
  278.         # 转静态路径，确保唯一性
  279.         key_name = os.path.abspath(key_name)
  280.         if IMSGetDatasBase.get_data_timeout_dict[key_name]:
  281.             IMSGetDatasBase.time_dict[key_name] = datetime.datetime.today()
  282.         if self.vars_helper is not None :
  283.             IMSGetDatasBase.time_dict[key_name] = datetime.datetime.today()
  284.             self.vars_helper.set_value('IMSGetDatasBase.time_list', IMSGetDatasBase.time_dict)
  285.     
  286.     def get_lock(self, key_name) -> threading.Lock:
  287.         '''获取锁'''
  288.         # 转静态路径，确保唯一性
  289.         key_name = os.path.abspath(key_name)
  290.         if key_name not in IMSGetDatasBase.lock_dict.keys():
  291.             IMSGetDatasBase.lock_dict[key_name] = threading.Lock()
  292.         return IMSGetDatasBase.lock_dict[key_name]
  294.     def get_data_of_date(
  295.         self,
  296.         save_dir,
  297.         sql,
  298.         sort_columns:List[str],
  299.         del_index_list=[-1], # 删除最后下标
  300.         start_date = datetime.datetime(2017, 1, 1), # 开始时间
  301.         offset = relativedelta(months=3), # 时间间隔
  302.         date_format_fun = lambda d: '%04d%02d01' % (d.year, d.month), # 查询语句中替代时间参数的格式化
  303.         filename_format_fun = lambda d: '%04d%02d.csv' % (d.year, d.month), # 查询语句中替代时间参数的格式化
  304.         stop_date = '20700101', # 超过时间则停止
  305.         ):
  306.         '''分时间增量读取数据'''
  307.         # 创建文件夹
  308.         if not os.path.exists(save_dir):
  309.             os.makedirs(save_dir)
  310.         else:
  311.             #删除最后一个文件
  312.             file_list = os.listdir(save_dir)
  313.             if len(file_list)>0:
  314.                 file_list.sort()
  315.                 for del_index in del_index_list:
  316.                     os.remove(os.path.join(save_dir,file_list[del_index]))
  317.                     print('删除最后一个文件：', file_list[del_index])
  318.         select_index = -1
  319.         # start_date = datetime.datetime(2017, 1, 1)
  320.         while True:
  321.             end_date = start_date + offset
  322.             start_date_str = date_format_fun(start_date)
  323.             end_date_str = date_format_fun(end_date)
  324.             self.logger.info('date: %s-%s', start_date_str, end_date_str)
  325.             file_path = os.path.join(save_dir, filename_format_fun(start_date))
  326.             # self.logger.info('file_path: %s', file_path)
  327.             if not os.path.exists(file_path):
  328.                 data:pd.DataFrame = self.db_helper.get_data(sql % (start_date_str, end_date_str))
  329.                 if data is None:
  330.                     break
  331.                 self.logger.info('data: %d', len(data))
  332.                 # self.logger.info('data: %d', data.columns)
  333.                 if len(data)>0:
  334.                     select_index+=1
  335.                     # 排序
  336.                     data = data.sort_values(sort_columns)
  337.                     data.to_csv(file_path)
  338.                 elif select_index!=-1:
  339.                     break
  340.                 elif stop_date < start_date_str:
  341.                     raise Exception("读取数据异常，时间超出最大值！")
  342.             start_date = end_date
  343. pass
  344. class CustomIMSGetDatas(IMSGetDatasBase):
  346.     def __init__(
  347.         self,
  348.         host='192.168.13.134',
  349.         port=4000,
  350.         database='test_ims',
  351.         user='root',
  352.         password='rootimmsadmin',
  353.         save_dir='./hjx/data/export_ims_data',
  354.         logger:logging.Logger=None
  355.         ):
  356.         self.save_dir = save_dir
  357.         self.logger = logger
  358.         super().__init__(
  359.             host=host,
  360.             port=port,
  361.             database=database,
  362.             user=user,
  363.             password=password,
  364.             save_dir=save_dir,
  365.             logger=logger
  366.             )
  368.     def get_ims_w_amt_pro(self):
  369.         '''年月系列占比数据'''
  370.         file_path = os.path.join(self.save_dir,'ims_w_amt_pro.csv')
  371.         now_lock = self.get_lock(file_path)
  372.         now_lock.acquire() # 加锁
  373.         try:
  374.             # 设置超时4小时才重新查数据
  375.             # if not self.get_last_time(file_path):
  376.             #     return
  377.         
  378.             sql = 'SELECT * FROM ims_w_amt_pro'
  379.             data:pd.DataFrame = self.db_helper.get_data(sql)
  380.             data = data.rename(columns={
  381.                 'serial_forecast_proportion': 'forecast_proportion',
  382.             })
  383.             data.to_csv(file_path)
  384.             # # 更新超时时间
  385.             # self.save_last_time(file_path)
  386.         except Exception as ex:
  387.             self.logger.exception(ex)
  388.             raise ex # 往外抛出异常
  389.         finally:
  390.             now_lock.release() # 释放锁
  391. pass
  392. def get_datas(
  393.     host='192.168.13.134',
  394.     port=4000,
  395.     database='test_ims',
  396.     user='root',
  397.     password='rootimmsadmin',
  398.     save_dir='./hjx/data/export_ims_data',
  399.     logger:logging.Logger=None
  400.     ):
  401.     ur_bi_get_datas = CustomIMSGetDatas(
  402.         host=host,
  403.         port=port,
  404.         database=database,
  405.         user=user,
  406.         password=password,
  407.         save_dir=save_dir,
  408.         logger=logger
  409.     )
  410.     try:
  411.         # 年月系列占比数据
  412.         logger.info('正在查询年月系列占比数据...')
  413.         ur_bi_get_datas.get_ims_w_amt_pro()
  414.         logger.info('查询年月系列占比数据完成！')
  415.         
  416.     except Exception as ex:
  417.         logger.exception(ex)
  418.         raise ex # 往外抛出异常
  419.     finally:
  420.         ur_bi_get_datas.close()
  421. pass
  422. def getdata_export_ims(
  423.     host='192.168.13.134',
  424.     port=4000,
  425.     database='test_ims',
  426.     user='root',
  427.     password='rootimmsadmin',
  428.     save_dir='./hjx/data/export_ims_data',
  429.     logger:logging.Logger=None
  430.     ):
  431.     get_datas(
  432.         host=host,
  433.         port=port,
  434.         database=database,
  435.         user=user,
  436.         password=password,
  437.         save_dir=save_dir,
  438.         logger=logger
  439.     )
  440. pass
标签： 数据库 hive 大数据
本文转载自: https://blog.csdn.net/my_name_is_learn/article/details/129243879
版权归原作者 是猪哥不是诸葛 所有， 如有侵权，请联系我们删除。
登录后发布评论

“实际业务读取Hive数据库（2023年2月）”的评论:

还没有评论
关于作者
...
overfit同步小助手
文章同步
相关阅读

网络安全法-网络运行安全

使用selenium/drissionpage时如何阻止chrome自动跳转http到https

docker desktop 里部署的Open WebUI 管理员密码忘记了的处理方法

在ubuntu20.04中搭建onsite比赛运行环境

利用开源的低代码表单设计器FcDesigner高效管理和渲染复杂表单结构

Kafka学习笔记

【前端】浏览器输入url到页面呈现发生了什么？