start_mysql.py 4.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147
  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):
  37. sqls = []
  38. for i in points:
  39. meter,funcid = i[0],i[1]
  40. print("%s开始查询%s至%s的数据 %s-%s"%(table,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,my_table):
  51. print("开始往mysql插入%s数据..."%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条..." % (my_table,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 = data["building"]["id"]
  74. my_database = data["mysql"]["database"]
  75. my_fjd_table = data["mysql"]["fjd_table"]
  76. my_energy_table = data["mysql"]["energy_table"]
  77. my_co2_table = data["mysql"]["co2_table"]
  78. my_pm25_table = data["mysql"]["pm25_table"]
  79. my_temperature_table = data["mysql"]["temperature_table"]
  80. my_hcho_table = data["mysql"]["hcho_table"]
  81. my_humidity_table = data["mysql"]["humidity_table"]
  82. mysql = data["mysql"]
  83. del mysql["fjd_table"]
  84. del mysql["energy_table"]
  85. del mysql["co2_table"]
  86. del mysql["pm25_table"]
  87. del mysql["temperature_table"]
  88. del mysql["hcho_table"]
  89. del mysql["humidity_table"]
  90. tables = ["fjd_0_near_15min","data_servicedata_15min"]
  91. # #连接hbase
  92. zillionUtil = ZillionUtil(url)
  93. #处理点位表
  94. pointlist = get_pointlist(hbase_database,"dy_pointlist",building)
  95. points = []
  96. for point in pointlist:
  97. i = point[1]
  98. if i == args.funcid:
  99. points.append(point)
  100. #连接mysql
  101. conn = pymysql.connect(**mysql)
  102. mysql_cur = conn.cursor()
  103. #电
  104. if args.funcid == 10101:
  105. sqls = hbase_energy_data(points, "fjd_0_near_15min")
  106. insert_mysql(sqls, my_fjd_table)
  107. sqls = hbase_energy_data(points, "data_servicedata_15min")
  108. insert_mysql(sqls, my_energy_table)
  109. #CO2
  110. if args.funcid == 11301:
  111. sqls = hbase_energy_data(points, "fjd_0_near_15min")
  112. insert_mysql(sqls, my_co2_table)
  113. #pm2.5
  114. if args.funcid == 11401:
  115. sqls = hbase_energy_data(points, "fjd_0_near_15min")
  116. insert_mysql(sqls, my_pm25_table)
  117. #甲醛
  118. if args.funcid == 11305:
  119. sqls = hbase_energy_data(points, "fjd_0_near_15min")
  120. insert_mysql(sqls, my_temperature_table)
  121. #温度
  122. if args.funcid == 11101:
  123. sqls = hbase_energy_data(points, "fjd_0_near_15min")
  124. insert_mysql(sqls, my_hcho_table)
  125. #湿度
  126. if args.funcid == 11201:
  127. sqls = hbase_energy_data(points, "fjd_0_near_15min")
  128. insert_mysql(sqls, my_humidity_table)
  129. mysql_cur.close()
  130. conn.close()