将json数据导入到ES集群——解决方案对比&填坑日记-摩杜云开发者社区

需求

将写好的json数据。导入到es集群

数据说明

文件JSON数据，一行一个JSON。

{"id":"d2716ae8fba4e026c4bd9445c3f49e2c","lang":"zh","title":"吉美旅馆","content":"吉美..."}
{"id":"d2716ae8fba4e026c4bd9445c3f49e2d","lang":"zh","title":"大话西游","content":"大话西游..."}

背景说明

ES 版本 8.X，开启了密码认证

使用elasticsearchDump 工具把json数据导入到ES集群中。

遇到的坑与解决方法

方案1：dump工具

工具：https://github.com/elasticsearch-dump/elasticsearch-dump

坑1：由于ES密码中有特殊符号。在使用dump工具的时候。命令如下

elasticdump \
  --input=/u01/isi/data/baike_result/1/baike_info_html_9.jsonl_step2 \
  --output="http://elastic:abcd%12%34@localhost:9200/self_index"

报错如下：URI malformed

Mon, 06 Nov 2023 11:59:20 GMT | starting dump
Mon, 06 Nov 2023 11:59:20 GMT | got 100 objects from source file (offset: 0)
Mon, 06 Nov 2023 11:59:20 GMT | Error Emitted => URI malformed
Mon, 06 Nov 2023 11:59:20 GMT | Error Emitted => URI malformed
Mon, 06 Nov 2023 11:59:20 GMT | Total Writes: 0
Mon, 06 Nov 2023 11:59:20 GMT | dump ended with error (get phase) => URIError: URI malformed

导致此问题的原因：因为密码中有特殊符号。

解决方案看这里：elasticsearch - How to solve malformed URI while using elasticdump? - Stack Overflow

方案里边说：需要把密码转ascll码。但是我测了没用，可能是因为百分号导致的。

我的最终解决方案是，在ES中新加了一个用户，不添加特殊符号。这种靠谱一些。但是操作起来，如果不熟悉ES的鉴权，也是很麻烦的。

Security APIs | Elasticsearch Guide [8.10] | Elastic这里是ES的添加用户的文档。

注意先添加权限，再配置角色，在添加用户。权限赋给角色，角色赋给用户。一通操作下来，还是遇到了问题，最终没能执行下去。

报错如下

status: 500,
  error: {
    type: 'not_x_content_exception',
    reason: 'not_x_content_exception: Compressor detection can only be called on some xcontent bytes or compressed xcontent bytes'
  }

问题原因。这里其实是最根本的原因。可以再看看上边的JSON数据。看起来没有问题，但是这种数据无法被dump工具识别。

我把数据改造了一下，如下所示

{"_index":"my_index","_id":"1","_source":{"id":"d2716ae8fba4e026c4bd9445c3f49e2c","lang":"zh","title":"吉美旅馆","content":"吉美..."}}

上边改造后的数据是可以写入的。但是需要做数据改造。也挺麻烦。

elasticdump \
  --input=/u01/isi/data/baike_result/1/temp.json \
  --output=http://elasticinsert:angus123321@localhost:9200/baike_info_test \
  --type=data

方案2：logstash工具

转弯掉头。因为改造数据也挺麻烦。特别是在大批量数据下。我决定使用logstash，第一不用决绝用户密码问题。第二不用处理数据

操作步骤如下：

第一步：下载logstash。下载地址：Logstash 8.8.0 | Elastic

第二步：上传服务器，并解压。

第三步：在解压后的conf目录下。添加一个logstash的配置 logstash-self.conf

input {
  file {
  #  path => "/u01/isi/data/result/1/baike_info_html_8.jsonl_step2"
    # 这里，我是获取指定目录下，全部的文件
    path => ["/u01/isi/data/result/2/*"]
    start_position => "beginning"
    sincedb_path => "/dev/null"
    codec => "json"
  }
}

filter {
  # 这里可以定义其他过滤规则。这里我只要json中的这几个字段。
  prune {
    whitelist_names => ["id", "url", "content", "title", "lang"]
  }
}

output {
  elasticsearch {
    hosts => "http://localhost:9200"
    index => "self_index"
    user => "elastic"
    password => "abcd%12%34"
    document_id => "%{id}"
  }
  # 这里是调试用的。可以看看json数据是否正确。
  # stdout {
  #  codec => rubydebug
  #}
}

第四步：启动开始做数据导入

到解压后的目录中，可以看到bin目录和conf目录。以后台的方式启动任务。日志会打印到当前目录中的 out.log 文件中。
 nohup bin/logstash -f  ./conf/logstash-self.conf   >> out.log &
 
 如果想结束任务。则使用 ps -ef |grep logstash 即可看到任务进程，使用kill -9 任务id即可结束任务。