澳门皇冠金沙网站▎在线官网
做最好的网站

导入elasticsearch方法总结澳门皇冠金沙网站,在线

2020-04-21 作者:网络服务   |   浏览(85)

背景

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

navicat是mysql可视化工具中最棒的,但是,在处理视图的导入导出方面,它是按照视图名称的字母顺序来处理的,若视图存在依赖,在导入过程中就会报错。前面已经用python写了一个,但在使用过程中,遇到xfffd编码,python的pymysql会直接崩溃。发现golang没有这个问题,正好用go重写,来熟悉golang。

2. 导elasticsearch参数对应名称

一些关键点

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

map & json,在处理主键与外键信息时,需要用到json数据结构来存储中间结果,因为要灵活处理,在golang中只能用map[string]interface{}来处理。 interface{} 相当于java中的object,能接受任意数据类型,方便但在使用时要做到心中有数,不然一旦数据类型不匹配,程序就会崩溃。 xfffd ,utf8中的占位符,超出范围的utf8mb4入库后,会被存储为xfffd,数据导出时,需要过滤掉。 goroutine, golang的并发支持很独特,我们的工具支持多个库同时备份,很容易使用goroutine来实现并行。

关系型数据库对应参数:
  服务器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 '{}'

main.go,并发、命令行参数

创建映射
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"
    }
}'

package common

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"
    }
}'

type OpFlag struct { Tables bool //表结构 Datum bool //表结构和数据 Views bool //视图 Funcs bool //函数与存储过程}
6. csv或txt格式数据导入到elasticsearch

注意事项:

  1. 如果文件中的第一行未设置字段名称可用下方命令行添加,或者在导入命令中修改指定参数添加
sed -i '1 s/^/username,email,passwordn/' user.txt   #为user.txt首行添加字段名username,email,password
  1. 在导入过程中,要导入的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插件执行上面的命令

澳门皇冠金沙网站,在线官网 1

使用head插件导入csv数据


main.go,程序入口,处理命令行参数

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
 if len(os.Args)  1 { flag = common.OpFlag{ Tables: false, Datum: false, Views: false, Funcs: false, } switch os.Args[1] { //接受一个参数 case "table": flag.Tables = true //根据参数设定标识量 case "data": flag.Tables = true flag.Datum = true case "views": flag.Views = true case "funcs": flag.Funcs = true default: //参数不正确,报错退出 log.Fatal("You arg must be in : table, data, views or funcs.") } }else{ //无参数,默认导出所有 flag = common.OpFlag{ Tables: true, Datum: true, Views: true, Funcs: true, } } err := backUp.Export(flag) 根据参数进行数据库备份
使用方法:

澳门皇冠金沙网站,在线官网 2

json2es.py使用方法

Export.go

备份主流程,根据configs.json生成goroutine来备份数据库,并等待完成。

本文由澳门皇冠金沙网站发布于网络服务,转载请注明出处:导入elasticsearch方法总结澳门皇冠金沙网站,在线

关键词: