快捷導(dǎo)航

python 解決動態(tài)的定義變量名,并給其賦值的方法(大數(shù)據(jù)處理)

更新時間：2018年11月10日 09:21:08 作者：威震四海

今天小編就為大家分享一篇python 解決動態(tài)的定義變量名,并給其賦值的方法(大數(shù)據(jù)處理)，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧

最近消費(fèi)kafka數(shù)據(jù)到磁盤的時候遇到了這樣的問題：

需求：每天大概有1千萬條數(shù)據(jù)，每條數(shù)據(jù)包含19個字段信息，需要將數(shù)據(jù)寫到服務(wù)器磁盤，以第二個字段作為大類建立目錄，第7個字段作為小類配合時間戳作為文件名，臨時文件后綴tmp，當(dāng)每個文件的寫入條數(shù)（可配置，比如100條）達(dá)到要求條數(shù)時，將后綴tmp改為out。

問題：大類共有30個，小類不計(jì)其數(shù)而且未知，比如大類為A,小類為a，時間戳為20180606095835234，則A目錄下的文件名為20180606095835234_a.tmp,這樣一來需要在此文件寫滿100條時，更新時間戳生成第二個文件名，如果此時有1000個文件都在寫則需要有1000個時間戳，和1000個計(jì)數(shù)器記錄每個文件當(dāng)前的條數(shù)，如果分別定義1000個變量顯然是不劃算的，

嘗試：中間過程想到了動態(tài)定義變量名，即

定義第七個字段：seven = data.split('|')[7]

定義文件名：filename = time_stamp + '_' + seven+'.tmp'，

定義文件計(jì)數(shù)器：seven + ‘_num' = 0

定義文件時間戳：seven + '_stamp' = time.time( )

想法其實(shí)是沒問題的，但是這里用到了一個不常用的語法：用一個變量名和一個字符串拼接出來一個新的變量名，并繼續(xù)賦值（不知道我的表述是否清楚），試過了用local（）函數(shù)、global（）函數(shù)、exec（）函數(shù)都沒有達(dá)到預(yù)期效果，也許是把問題想的太復(fù)雜了

解決：最后使用三個字典將這個問題完美解決，

定義一個字典用來存計(jì)數(shù)器，字典的每一個鍵對應(yīng)一個文件名，值對應(yīng)當(dāng)前計(jì)數(shù)，并實(shí)時更新；

定義一個字典用來存時間戳，鍵對應(yīng)一個文件名，值對應(yīng)時間戳，達(dá)到100條就更新一次；

定義一個字典用來存大類，鍵對應(yīng)代號，值對應(yīng)分類；

局部功能代碼如下：

def kafka_to_disk():
 print('啟動前檢測上次運(yùn)行時是否存在意外中斷的數(shù)據(jù)文件......')
 print('搜索最近一次執(zhí)行腳本產(chǎn)生的時間目錄......')
 # 待處理臨時文件列表
 tmp_list = []
 try:
  for category_dir in os.listdir(local_file_path):
   if len(os.listdir(local_file_path+os.sep+category_dir)) > 0:
    for file in os.listdir(local_file_path+os.sep+category_dir):
     if suffix in file:
      tmp_list.append(local_file_path+os.sep+category_dir+os.sep+file)
  # print('上次運(yùn)行程序產(chǎn)生的臨時文件有---{}'.format(tmp_list))
 except Exception as e:
  pass
 if len(tmp_list) == 0:
  print('未掃描任何殘留臨時文件')
 else:
  print('開始修復(fù)殘留臨時文件......')
 tmp_num = 0
 for tmp in tmp_list:
  os.rename(tmp, tmp.split('.')[0]+'.out')
  tmp_num += 1
 print('本次啟動共修復(fù)殘留臨時文件★★★★★-----{}個-----★★★★★'.format(tmp_num))
 
 category_poor = {
  '1': 'news', '2': 'weibo', '3': 'weixin', '4': 'app', '5': 'newspaper', '6': 'luntan',
  '7': 'blog', '8': 'video', '9': 'shangji', '10': 'shangjia', '11': 'gtzy', '12': 'zfztb',
  '13': 'gyfp', '14': 'gjz', '15': 'zfxx', '16': 'ptztb', '17': 'company', '18': 'house',
  '19': 'hospital', '20': 'bank', '21': 'zone', '22': 'express', '23': 'zpgw', '24': 'zscq',
  '25': 'hotel', '26': 'cpws', '27': 'gxqy', '28': 'gpjj', '29': 'dtyy', '30': 'bdbk'}
 
 time_stamp = utils.get_time_stamp() # 初始化毫秒級時間戳 ： 20180509103015125
 consumer = KafkaConsumer(topic, group_id=group_id, auto_offset_reset=auto_offset_reset, bootstrap_servers=eval(bootstrap_servers))
 print('連接kafka成功,數(shù)據(jù)篩選中......')
 file_poor = {}       # 子類池用于文件計(jì)數(shù)器
 time_stamp_poor = {}     # 子類時間戳池，用于觸發(fā)文件切換
 time_stamp = utils.get_time_stamp()  # 初始化毫秒級時間戳 ：20180509103015125
 for message in consumer:
  # 提取第8個字段自動匹配目錄進(jìn)行創(chuàng)建
  if message.value.decode().split('|')[1] in category_poor:
   category = category_poor[message.value.decode().split('|')[1]]
  else:
   print(message.value.decode())
   continue
  category_dir = local_file_path + os.sep + category
  if not os.path.exists(category_dir):
   os.makedirs(category_dir)
  # 提取第2個字段，用于生成文件名
  if message.value.decode().split('|')[7] in time_stamp_poor:
   shot_file_name = time_stamp_poor[message.value.decode().split('|')[7]] + '_' + message.value.decode().split('|')[7]
  else:
   shot_file_name = time_stamp + '_' + message.value.decode().split('|')[7]
  file_name = category_dir + os.sep + shot_file_name + '.tmp'
 
  # 給每一個文件設(shè)定一個計(jì)數(shù)器
  if message.value.decode().split('|')[7] not in file_poor:
   file_poor[message.value.decode().split('|')[7]] = 0
 
  with open(file_name, 'a', encoding='utf-8')as f1:
   f1.write(message.value.decode())
   file_poor[message.value.decode().split('|')[7]] += 1
 
  # 觸發(fā)切換文件的操作,用時間戳生成第二文件名
  if file_poor[message.value.decode().split('|')[7]] == strip_number:
   time_stamp_poor[message.value.decode().split('|')[7]] = utils.get_time_stamp()
   file_poor[message.value.decode().split('|')[7]] = 0

以上這篇python 解決動態(tài)的定義變量名,并給其賦值的方法(大數(shù)據(jù)處理)就是小編分享給大家的全部內(nèi)容了，希望能給大家一個參考，也希望大家多多支持腳本之家。

您可能感興趣的文章: