我从https://www.n2yo.com/api/流式传输RESTful API数据,用于跟踪卫星位置。我在Elasticsearch中使用python客户端。我每10秒将流数据保存到ES,并通过Kibana进行可视化。我的ES分数是6.4.3
我的代码是:
URL = "https://www.n2yo.com/rest/v1/satellite/positions/25544/41.702/-76.014/0/2/&apiKey= your key"
es = Elasticsearch('http://ip:port',timeout=600)
settings = { "settings": {
"number_of_shards":1,
'number_of_replicas':0
},
"mappings" : {
"document" : {
"properties":{
"geo": {
"type": "geo_point"
}
}
}
}
}
try:
es.indices.create(index = "spacestation", body=settings)
except RequestError as es1:
print('Index already exists!!')
sys.exit(1)
def collect_data():
data = requests.get(url = URL).json()
del data['positions'][1]
new_data = {'geo':{'lat':data['positions'][0]['satlatitude'],
'lon':data['positions'][0]['satlongitude']},
'satname': data['info']['satname'], 'satid': data['info']['satid'],
'timestamp':datetime.fromtimestamp(data['positions'][0]['timestamp']).isoformat()
}
es.index(index='spacestation', doc_type='document', body=new_data)
schedule.every(10).seconds.do(collect_data)
while True:
schedule.run_pending()
time.sleep(1)
我的问题是:昨天我失去了联系。错误如下,
requests.exceptions.ConnectionError:HTTPSConnectionPool(host ='www.n2yo.com',port = 443):URL超过了最大重试次数:/rest/v1/satellite/positions/25544/41.702/-76.014/0/2 /&apiKey =(由NewConnectionError(':导致无法建立新连接:[Errno -3]名称解析临时失败',))
重新运行代码时,我不能,因为索引已存在。如果删除索引,我将丢失ES中已经存在的数据。我可以做什么?我需要保留保存的数据,并且从现在开始运行作业。有什么解决办法吗?
仅当您从n2yo.com接收数据时,才创建索引。您应该使用函数es.indices.exists
。然后在失败的情况下使函数collect_data
递归。试试:
URL = "https://www.n2yo.com/rest/v1/satellite/positions/25544/41.702/-76.014/0/2/&apiKey= your key"
es = Elasticsearch('http://ip:port',timeout=600)
def create_index()
if not es.indices.exists(index = "spacestation"):
settings = { "settings": {
"number_of_shards":1,
'number_of_replicas':0
},
"mappings" : {
"document" : {
"properties":{
"geo": {
"type": "geo_point"
}
}
}
}
}
es.indices.create(index = "spacestation", body=settings)
else:
print('Index already exists!!')
def collect_data():
try:
data = requests.get(url = URL).json()
create_index()
del data['positions'][1]
new_data = {'geo':{'lat':data['positions'][0]['satlatitude'],
'lon':data['positions'][0]['satlongitude']},
'satname': data['info']['satname'], 'satid': data['info']['satid'],
'timestamp':datetime.fromtimestamp(data['positions'][0]['timestamp']).isoformat()
}
es.index(index='spacestation', doc_type='document', body=new_data)
except:
collect_data()
schedule.every(10).seconds.do(collect_data)
while True:
schedule.run_pending()
time.sleep(1)