99 lines
3.2 KiB
Python
99 lines
3.2 KiB
Python
|
#!/usr/bin/env python3
|
|||
|
|
# coding: utf-8
|
|||
|
|
# File: insert_es.py
|
|||
|
|
# Author: lhy<lhy_in_blcu@126.com,https://huangyong.github.io>
|
|||
|
|
# Date: 18-10-10
|
|||
|
|
|
|||
|
|
import os
|
|||
|
|
import time
|
|||
|
|
|
|||
|
|
import json
|
|||
|
|
from elasticsearch import Elasticsearch
|
|||
|
|
from elasticsearch.helpers import bulk
|
|||
|
|
import pymongo
|
|||
|
|
|
|||
|
|
class ProcessIntoES:
|
|||
|
|
def __init__(self):
|
|||
|
|
self._index = "crime_data"
|
|||
|
|
self.es = Elasticsearch([{"host": "127.0.0.1", "port": 9200}])
|
|||
|
|
self.doc_type = "crime"
|
|||
|
|
cur = '/'.join(os.path.abspath(__file__).split('/')[:-1])
|
|||
|
|
self.music_file = os.path.join(cur, 'qa_corpus.json')
|
|||
|
|
|
|||
|
|
'''创建ES索引，确定分词类型'''
|
|||
|
|
def create_mapping(self):
|
|||
|
|
node_mappings = {
|
|||
|
|
"mappings": {
|
|||
|
|
self.doc_type: { # type
|
|||
|
|
"properties": {
|
|||
|
|
"question": { # field: 问题
|
|||
|
|
"type": "text", # lxw NOTE: cannot be string
|
|||
|
|
"analyzer": "ik_max_word",
|
|||
|
|
"search_analyzer": "ik_smart",
|
|||
|
|
"index": "true" # The index option controls whether field values are indexed.
|
|||
|
|
},
|
|||
|
|
"answers": { # field: 问题
|
|||
|
|
"type": "text", # lxw NOTE: cannot be string
|
|||
|
|
"analyzer": "ik_max_word",
|
|||
|
|
"search_analyzer": "ik_smart",
|
|||
|
|
"index": "true" # The index option controls whether field values are indexed.
|
|||
|
|
},
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
if not self.es.indices.exists(index=self._index):
|
|||
|
|
self.es.indices.create(index=self._index, body=node_mappings)
|
|||
|
|
print("Create {} mapping successfully.".format(self._index))
|
|||
|
|
else:
|
|||
|
|
print("index({}) already exists.".format(self._index))
|
|||
|
|
|
|||
|
|
'''批量插入数据'''
|
|||
|
|
def insert_data_bulk(self, action_list):
|
|||
|
|
success, _ = bulk(self.es, action_list, index=self._index, raise_on_error=True)
|
|||
|
|
print("Performed {0} actions. _: {1}".format(success, _))
|
|||
|
|
|
|||
|
|
|
|||
|
|
'''初始化ES，将数据插入到ES数据库当中'''
|
|||
|
|
def init_ES():
|
|||
|
|
pie = ProcessIntoES()
|
|||
|
|
# 创建ES的index
|
|||
|
|
pie.create_mapping()
|
|||
|
|
start_time = time.time()
|
|||
|
|
index = 0
|
|||
|
|
count = 0
|
|||
|
|
action_list = []
|
|||
|
|
BULK_COUNT = 1000 # 每BULK_COUNT个句子一起插入到ES中
|
|||
|
|
|
|||
|
|
for line in open(pie.music_file):
|
|||
|
|
if not line:
|
|||
|
|
continue
|
|||
|
|
item = json.loads(line)
|
|||
|
|
index += 1
|
|||
|
|
action = {
|
|||
|
|
"_index": pie._index,
|
|||
|
|
"_type": pie.doc_type,
|
|||
|
|
"_source": {
|
|||
|
|
"question": item['question'],
|
|||
|
|
"answers": '\n'.join(item['answers']),
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
action_list.append(action)
|
|||
|
|
if index > BULK_COUNT:
|
|||
|
|
pie.insert_data_bulk(action_list=action_list)
|
|||
|
|
index = 0
|
|||
|
|
count += 1
|
|||
|
|
print(count)
|
|||
|
|
action_list = []
|
|||
|
|
end_time = time.time()
|
|||
|
|
|
|||
|
|
print("Time Cost:{0}".format(end_time - start_time))
|
|||
|
|
|
|||
|
|
|
|||
|
|
if __name__ == "__main__":
|
|||
|
|
# 将数据库插入到elasticsearch当中
|
|||
|
|
# init_ES()
|
|||
|
|
# 按照标题进行查询
|
|||
|
|
question = '我老公要起诉离婚 我不想离婚怎么办'
|
|||
|
|
|