start_mysql.py 5.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152
  1. import json
  2. from MyUtils.ZillionUtil import ZillionUtil
  3. import pymysql
  4. import datetime
  5. import argparse
  6. INSERT_SQL = "replace into %s.%s(building,meter,func_id,data_time,data_value,dt) values "
  7. # 获取项目点位
  8. def get_pointlist(hbase_database, hbase_table, buildingid):
  9. Criteria = {
  10. "building": buildingid
  11. }
  12. datas = zillionUtil.select(hbase_database, hbase_table, Criteria)
  13. pointlist = []
  14. for i in datas:
  15. point = []
  16. meter = i["meter"]
  17. funcid = i["funcid"]
  18. point.append(meter)
  19. point.append(funcid)
  20. pointlist.append(point)
  21. return pointlist
  22. # 获取能耗数据
  23. def get_data_time(hbase_database, hbase_table, buildingid, meter, funcid,from_time,to_time):
  24. Criteria = {
  25. "building": buildingid,
  26. "meter": meter,
  27. "funcid": funcid,
  28. "data_time":{
  29. "$gte": from_time,
  30. "$lt": to_time
  31. }
  32. }
  33. datas = zillionUtil.select(hbase_database, hbase_table, Criteria)
  34. return datas
  35. #取hbase数据,处理成sql语句
  36. def hbase_energy_data(points,table,building):
  37. sqls = []
  38. for i in points:
  39. meter,funcid = i[0],i[1]
  40. print("%s开始查询项目%s的%s至%s的数据 %s-%s"%(table,building,args.start_time,args.end_time,meter,funcid))
  41. datas = get_data_time(hbase_database,table,building,meter,funcid,args.start_time,args.end_time)
  42. for i in datas:
  43. data_time = i["data_time"]
  44. data_value = i["data_value"]
  45. dt = data_time[0:4]+ "-" + data_time[4:6] + "-" + data_time[6:8]
  46. sqlline = "(%s,%s,%s,%s,%s,'%s')" % (building, meter, funcid, data_time, data_value, dt)
  47. sqls.append(sqlline)
  48. return sqls
  49. # mysql插入数据
  50. def insert_mysql(sqls,building,my_table):
  51. print("开始往mysql插入%s的%s数据..."%(building,my_table))
  52. for i in range(0, len(sqls), 1000):
  53. sqlranges = sqls[i:i + 1000]
  54. sqlranges = INSERT_SQL % (my_database,my_table) + ",".join(sqlranges)
  55. mysql_cur.execute(sqlranges)
  56. conn.commit()
  57. print("mysql数据%s,项目%s插入成功,合计%s条..." % (my_table,building,len(sqls)))
  58. end_time = datetime.datetime.now().strftime("%Y%m%d") + "000000"
  59. start_time = (datetime.datetime.now() + datetime.timedelta(days=-1)).strftime("%Y%m%d") + "000000"
  60. parser = argparse.ArgumentParser()
  61. parser.add_argument("--start_time",default=start_time,help="--start_time 20220101000000")
  62. parser.add_argument("--end_time",default=end_time,help="--end_time 20220102000000")
  63. parser.add_argument("funcid",type=int,help="--funcid 10101")
  64. args = parser.parse_args()
  65. # print(args.start_time)
  66. # print(args.end_time)
  67. # print(args.funcid)
  68. # 读取配置文件信息
  69. with open("config.json", "r") as f:
  70. data = json.load(f)
  71. hbase_database = data["metadata"]["database"]
  72. url = data["metadata"]["url"]
  73. building_list = data["building"]["id"]
  74. print("项目列表:%s"%building_list)
  75. my_database = data["mysql"]["database"]
  76. my_fjd_table = data["mysql"]["fjd_table"]
  77. my_energy_table = data["mysql"]["energy_table"]
  78. my_co2_table = data["mysql"]["co2_table"]
  79. my_pm25_table = data["mysql"]["pm25_table"]
  80. my_temperature_table = data["mysql"]["temperature_table"]
  81. my_hcho_table = data["mysql"]["hcho_table"]
  82. my_humidity_table = data["mysql"]["humidity_table"]
  83. mysql = data["mysql"]
  84. del mysql["fjd_table"]
  85. del mysql["energy_table"]
  86. del mysql["co2_table"]
  87. del mysql["pm25_table"]
  88. del mysql["temperature_table"]
  89. del mysql["hcho_table"]
  90. del mysql["humidity_table"]
  91. tables = ["fjd_0_near_15min","data_servicedata_15min"]
  92. # #连接hbase
  93. zillionUtil = ZillionUtil(url)
  94. #连接mysql
  95. conn = pymysql.connect(**mysql)
  96. mysql_cur = conn.cursor()
  97. for building in building_list:
  98. #处理点位表
  99. pointlist = get_pointlist(hbase_database,"dy_pointlist",building)
  100. if pointlist:
  101. points = []
  102. for point in pointlist:
  103. i = point[1]
  104. if i == args.funcid:
  105. points.append(point)
  106. #电
  107. if args.funcid == 10101:
  108. sqls = hbase_energy_data(points, "fjd_0_near_15min",building)
  109. insert_mysql(sqls,building, my_fjd_table)
  110. sqls = hbase_energy_data(points, "data_servicedata_15min",building)
  111. insert_mysql(sqls,building, my_energy_table)
  112. #CO2
  113. if args.funcid == 11301:
  114. sqls = hbase_energy_data(points, "fjd_0_near_15min",building)
  115. insert_mysql(sqls,building, my_co2_table)
  116. #pm2.5
  117. if args.funcid == 11401:
  118. sqls = hbase_energy_data(points, "fjd_0_near_15min",building)
  119. insert_mysql(sqls,building, my_pm25_table)
  120. #甲醛
  121. if args.funcid == 11305:
  122. sqls = hbase_energy_data(points, "fjd_0_near_15min",building)
  123. insert_mysql(sqls,building, my_hcho_table)
  124. #温度
  125. if args.funcid == 11101:
  126. sqls = hbase_energy_data(points, "fjd_0_near_15min",building)
  127. insert_mysql(sqls,building, my_temperature_table)
  128. #湿度
  129. if args.funcid == 11201:
  130. sqls = hbase_energy_data(points, "fjd_0_near_15min",building)
  131. insert_mysql(sqls, building,my_humidity_table)
  132. else:
  133. print("未查询到项目%s的点位表..."%building)
  134. mysql_cur.close()
  135. conn.close()