Easticsearch 数据迁移至influxdb【python】

wewqeqw

Easticsearch 数据迁移至influxdb python

需求：将Easticsearch部分数据迁移至influxdb中。

见过从mysql，influxdb迁移至Easticsearch中的，没见过从Easticsearch迁移至influxdb中，迁移的数据是一些实时性的流量数据，influxdb时序性数据库对这类数据的支撑比较客观。

解决方案：大批量从Easticsearch取数据，两种方案。1.from...size    2.scroll (类似于数据库的游标)  脚本采用第二种scroll方案对Easticsearch 查询取数据。循环通过scrool_id进行查询并写入influxdb中。

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92

#!/usr/bin/env python #coding=utf-8 import sys import json import datetime import elasticsearch from influxdb import InfluxDBClient #连接Easticsearch class ES(object):     @classmethod     def connect_host(cls):         url = "http://192.168.121.33:9202/"         es = elasticsearch.Elasticsearch(url,timeout=120)         return es es = ES.connect_host() #连接influxdb client = InfluxDBClient(host="192.168.121.33", port="8086", username='admin', password='admin', database='esl') client.create_database('esl') #DSL查询语法 data = {     "query": { "match_all" : {}},     "size": 100 } # 设置要过滤返回的字段值，要什么字段。     'hits.hits._source.resource_id',     'hits.hits._source.timestamp',     'hits.hits._source.counter_volume',     'hits.hits._source.@timestamp', ] # 指定search_type="scan"模式，并返回_scroll_id给es.scroll获取数据使用 res = es.search(     index='pipefilter_meters*',     doc_type ='canaledge.flow.bytes',     body=data,     search_type="scan",     scroll="10m" ) scroll_id = res['_scroll_id'] response= es.scroll(scroll_id=scroll_id, scroll= "10m",filter_path=return_fields,) scroll_id = response['_scroll_id']   #获取第二次scroll_id hits = response['hits']['hits'] in_data = [] while len(hits) > 0:     for i in hits:         res_id = i['_source']['resource_id']         r_id, r_type = res_id.split(':')         datas = {             "measurement": "es_net",             "tags": {                  "resource_id": r_id,                  "type": r_type              },             "time": i['_source']['timestamp'],             "fields": {                 "counter_volume": i['_source']['counter_volume']             }         }         in_data.append(datas)     #循环写入influxdb     client.write_points(in_data)     in_data = []   #每次循环完重新定义列表为空     data = {         "query": { "match_all" : {}},         "size": 100     }     ## 设置要过滤返回的字段值，要什么字段。         '_scroll_id',         'hits.hits._source.resource_id',         'hits.hits._source.timestamp',         'hits.hits._source.counter_volume',         'hits.hits._source.@timestamp',     ]     ## 指定search_type="scan"模式，并返回_scroll_id给es.scroll获取数据使用     response= es.scroll(scroll_id=scroll_id, scroll= "10m",filter_path=return_fields,)     #调试     #if not response.get('hits'):     #    print response     #    sys.exit(1)     #else:     hits = response['hits']['hits']     scroll_id = response["_scroll_id"] #获取第三次scroll_id

江南

在实际生产环境中有用着没，这个数据到数据库中，那这样后期产生的数据会自动写到数据库中不，还是说还的到elastric中。