导入elasticsearch方法总结

1. 所需驱动和工具

安装mssql和mysql的jdbc驱动请参考上一篇：elasticsearch环境搭建过程如果在windows下使用导elasticsearch的命令：推荐使用cygwin用kopf插件查看导数据进度，若导入中断或错误，也可使用kopf删除该索引，重新导入

---

2. 导elasticsearch参数对应名称

elasticsearch对应参数：

elasticsearch服务器IP：192.168.1.1 索引名称：indexname #可以将其看成关系型数据库中的数据库名称 类型名称：typename #可以将其看成关系型数据库中的表名称 映射：mapping #可以将其看成关系型数据库中的表结构，默认分词，可进行设置 elasticsearch端口号：9200 who_jdbc_river: 每次导入数据，此处不能相同

关系型数据库对应参数：

服务器IP：192.168.1.2 用户：此处分别以 mssql的sa 和 mysql的root举例 数据库名称:dbname 表名称/视图名:tbname 密码：dbpasswd

---

3. 根据需求创建映射

视图（mapping）相当于关系型数据库的表结构，在执行之后导入命令前最好先创建索引和mapping，设置好数据类型和是否分词等关键参数，以免以后调用数据或者使用数据时陷入困境，最后不得不重新导入。

创建索引indexname：

curl -XPUT 192.168.1.1:9200/indexname -d '{}'

创建映射

curl -XPUT 192.168.1.1:9200/indexname/typename/_mapping -d '{ "typename" : { "properties" : { "domain" : { #数据字段名称 "type" : "string", #指定类型 "index": "not_analyzed" #指定不分词 }, "record_type" : { "type" : "string", "index": "not_analyzed" }, "record_value" : { "type" : "string", "index": "not_analyzed" } } } }'

---

4. 从mysql数据库导入到elasticsearch

curl -XPUT '192.168.1.1:9200/_river/who_jdbc_river/_meta' -d '{ "type" : "jdbc", "shedule" : null, "jdbc" : { "url" : "jdbc:mysql://192.168.1.2:3306/dbname", "user" : "root", "password" : "dbpasswd", "sql" : "select * from tbname", "index" : "indexname", "type" : "typename" } }'

---

5. 从mssql数据库导入到elasticsearch

curl -XPUT '192.168.1.1:9200/_river/who_jdbc_river/_meta' -d ' { "type" : "jdbc", "jdbc": { "url":"jdbc:sqlserver://192.168.1.2:1433;databaseName=dbname", "user":"sa", "password":"dbpasswd", "sql":"select * from dbo.tbname", "index" : "indexname", "type" : "typename" } }'

---

6. csv或txt格式数据导入到elasticsearch

注意事项：

如果文件中的第一行未设置字段名称可用下方命令行添加，或者在导入命令中修改指定参数添加

sed -i '1 s/^/username,email,password\n/' user.txt #为user.txt首行添加字段名username,email,password

在导入过程中，要导入的csv文件的文件名 如user.csv 会变成user.csv.processing，导入成功后user.csv.processing.imported 如果导入过程中出现错误中途断开，在重新导数据前记得先将文件名user.csv.processing.imported 改成user.csv 否则会提示找不到文件

curl -XPUT 192.168.1.1:9200/_river/who_jdbc_river/_meta -d ' { "type" : "csv", #指定文件类型 "csv_file" : { "folder" : "//home//black3y//", #要导入的文件的文件路径，注意注释斜杠，window下：D://isc//b "filename_pattern" : "user.csv", #待导入数据文件名称（txt或csv），支持同类型所有文件（.*\\.csv$） "poll":"5m", "fields" : [ "username", "email", "password" ], "first_line_is_header" : "false", #true:将第一行作为字段名，false:将fields中的信息作为字段名 "field_separator" : "\t", #tab对应\t，逗号直接写, "field_id" : "id", "field_id_include" : "false", "concurrent_requests" : "1", "charset" : "UTF-8", "script_before_all": "/path/to/before_all.sh", "script_after_all": "/path/to/after_all.sh", "script_before_file": "/path/to/before_file.sh", "script_after_file": "/path/to/after_file.sh" }, "index" : { "index" : "indexname", "type" : "typename", "bulk_size" : 100, "bulk_threshold" : 10 } }'

也可使用head插件执行上面的命令

使用head插件导入csv数据

---

7. json格式数据导入elasticsearch

赠上现成好用的python脚本 ( json2es.py )

#!/usr/bin/python # -*- coding: UTF-8 -*- from itertools import islice import json , sys from elasticsearch import Elasticsearch , helpers import threading _index = 'indexname' #修改为索引名 _type = 'typename' #修改为类型名 es_url = 'http://192.168.1.1:9200/' #修改为elasticsearch服务器 reload(sys) sys.setdefaultencoding('utf-8') es = Elasticsearch(es_url) #es.indices.create(index='webinfo', ignore=400,body = mapping) es.indices.create(index=_index, ignore=400) chunk_len = 10 num = 0 def bulk_es(chunk_data): bulks=[] try: for i in xrange(chunk_len): bulks.append({ "_index": _index, "_type": _type, "_source": chunk_data[i] }) helpers.bulk(es, bulks) except: pass with open(sys.argv[1]) as f: while True: lines = list(islice(f, chunk_len)) num =num +chunk_len sys.stdout.write('\r' + 'num:'+'%d' % num) sys.stdout.flush() bulk_es(lines) if not lines: print "\n" print "task has finished" break

使用方法：

原文链接：http://www.jianshu.com/p/1bf53ebfcdcf