update

write_data_into_es/revise_data_from_es.py

@@ -7,7 +7,8 @@ Created on Thu Aug  9 16:47:12 2018
 
 import elasticsearch
 import datetime
-from elasticsearch_7.helpers import scan
+from elasticsearch.helpers import scan
+import elasticsearch_7 as e7
 # import pandas as pd
 import json
 # from func_cal_doc_id import cal_doc_id
@@ -16,12 +17,9 @@ import json
 
 hosts = '172.16.31.17'
 port = 9000
-# user = 'zhouyujiang'
-# passwd = '8tM9JDN2LVxM'
-# http_auth = (user, passwd)
 es2 = elasticsearch.Elasticsearch(hosts=hosts, port=port)
 
-es7 = elasticsearch.Elasticsearch(hosts="172.16.52.27", port=9200,http_auth=())
+es7 = e7.Elasticsearch(hosts="172.16.52.27", port=9200,http_auth=("elastic","gengmei!@#"))
 
 
 target_index = 'gm-dbmw-doctor-read'
@@ -30,76 +28,53 @@ target_type = 'doctor'
 from_index = 'gm-dbmw-doctor-read'
 from_type = 'doc'
 
-fn = r'C:\Users\zhouyujiang\cuowu3.csv'
-bulk_all_body = ''
-
 
+bulk_all_body = ''
 
 search_body = {
     "query": {
         "bool": {
             "filter": [
-                {"term": {"platform.keyword": platform}},
-                {"term": {"releaser.keyword": releaser}},
-                {"exists": {"field": "play_count"}},
-                {"range": {"release_time": {"gte": 1519833600000,
-                                            "lt": 1522512000000}}},
-
-                {"range": {"duration": {"lte": 600}}}
             ]
         }
     }
 }
-q3_re = es.search(index=target_index, doc_type=target_type,
+es2_re = es2.search(index=target_index, doc_type=target_type,
                   body=search_body)
-q3_total = q3_re['hits']['total']
-write_into_scan = scan(client=es,
+q3_total = es2_re['hits']['total']
+write_into_scan = scan(client=es2,
                        query=search_body,
                        index=target_index,
                        doc_type=target_type,
                        scroll='5m',
                        request_timeout=100
                        )
-for one_scan in write_into_scan:
-    have_id = one_scan['_id']
-    wirte_set.add(have_id)
-print(platform, releaser, 'start_have', len(wirte_set))
-#        search_body['query']['bool']['filter'].append({"range": {"fetch_time":
-#                                 {"gte": 1547539200000}}})
-scan_re = scan(client=es,
-               query=search_body,
-               index=from_index,
-               doc_type=from_type,
-               scroll='5m',
-               request_timeout=100
-               )
 
 count = 0
-set_url = set()
-for one_scan in scan_re:
-#            print(one_scan)
+for one_scan in write_into_scan:
     count = count + 1
     line = one_scan['_source']
-            bulk_head = '{"index": {"_id":"%s"}}' % doc_id
-            data_str = json.dumps(line, ensure_ascii=False)
-
-            bulk_one_body = bulk_head + '\n' + data_str + '\n'
-            #
-            bulk_all_body += bulk_one_body
-            if count%100 == 0:
-
-                 eror_dic=es.bulk(index=target_index, doc_type=target_type,
-                        body=bulk_all_body, request_timeout=200)
-                 bulk_all_body=''
-                 if eror_dic['errors'] is True:
-                     print(eror_dic['items'])
-                     print(bulk_all_body)
-                 print(count)
+    doc_id = one_scan['_id']
+    bulk_head = '{"index": {"_id":"%s"}}' % doc_id
+    data_str = json.dumps(line, ensure_ascii=False)
+
+    bulk_one_body = bulk_head + '\n' + data_str + '\n'
+    print(bulk_one_body)
+    bulk_all_body += bulk_one_body
+    break
+    if count%100 == 0:
+
+         eror_dic=es7.bulk(index=target_index,
+                body=bulk_all_body, request_timeout=200)
+         bulk_all_body=''
+         if eror_dic['errors'] is True:
+             print(eror_dic['items'])
+             print(bulk_all_body)
+         print(count)
 
 if bulk_all_body != '':
-     eror_dic = es.bulk(body=bulk_all_body,
+     eror_dic = es7.bulk(body=bulk_all_body,
                         index=target_index,
-                        doc_type=target_type ,
                         request_timeout=200)
      if eror_dic['errors'] is True:
          print(eror_dic)