# -*- coding: utf-8 -*- """ Created on Thu Aug 9 16:47:12 2018 @author: zhouyujiang """ import elasticsearch import datetime from elasticsearch_7.helpers import scan # import pandas as pd import json # from func_cal_doc_id import cal_doc_id # from urllib import parse 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=()) target_index = 'gm-dbmw-doctor-read' target_type = 'doctor' from_index = 'gm-dbmw-doctor-read' from_type = 'doc' fn = r'C:\Users\zhouyujiang\cuowu3.csv' 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, body=search_body) q3_total = q3_re['hits']['total'] write_into_scan = scan(client=es, 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) 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) if bulk_all_body != '': eror_dic = es.bulk(body=bulk_all_body, index=target_index, doc_type=target_type , request_timeout=200) if eror_dic['errors'] is True: print(eror_dic) bulk_all_body = ''