2. 程式人生 > >Elasticsearch 批量匯入資料

Elasticsearch 批量匯入資料

阿新 • • 發佈：2019-01-04

前言

 可以非常方便地進行資料的多維分析，所以大資料分析領域也經常會見到它的身影，生產環境中絕大部分新產生的資料可以通過應用直接匯入，但是歷史或初始資料可能會需要單獨處理，這種情況下可能遇到需要匯入大量資料的情況

這裡簡單分享一下批量匯入資料的操作方法與相關基礎，還有可能會碰到的問題，詳細內容可以參考 官方文件

Tip: 當前的最新版本為 Elasticsearch 2.2.0

概要

bulk API

ES提供了一個叫 bulk 的 API 來進行批量操作

它用來在一個API呼叫中進行大量的索引更新或刪除操作，這極大的提升了操作效率

形式

API

API 可以是 /_bulk, /{index}/_bulk, 或 {index}/{type}/_bulk

 這三種形式，當索引或型別已經指定後，資料檔案中如不明確指定或申明的內容，就會預設使用API中的值

API 以是 /_bulk 結尾的，並且跟上如下形式的 JSON 資料

資料內容格式

action_and_meta_data\n
optional_source\n
action_and_meta_data\n
optional_source\n
....
action_and_meta_data\n
optional_source\n

Note: 最後的一行也必須以 \n 結尾

可用方法

可用的操作有 index, create, delete 和 update ：

  • index
     和 create 得在操作與元資料(action_and_meta_data)之後另起一行然後接上內容(必須遵循這樣的格式 ，後面會演示不這麼做導致操作失敗的示例)
  • delete 只用接上元資料就可以了，不必接上內容(原因自不用說，定位到文件就OK了)
  • update 得接上要變更的區域性資料，也得另起一行

文字指定

由於是批量操作，所以不太會直接使用命令列的方式手動指定，更多的是使用檔案，如果使用文字檔案，則得遵循如下格式

curl -s -XPOST localhost:9200/_bulk --data-binary "@requests"

Tip: requests 是檔名 , -s 是靜默模式，不產生輸出，也可以使用 > /dev/null

 替代

匯入資料

嘗試不按要求索引資料

[root@es-bulk tmp]# curl localhost:9200/stuff_orders/order_list/903713?pretty
{
  "_index" : "stuff_orders",
  "_type" : "order_list",
  "_id" : "903713",
  "found" : false
}
[root@es-bulk tmp]# cat test.json 
{"index":{"_index":"stuff_orders","_type":"order_list","_id":903713}}{"real_name":"劉備","user_id":48430,"address_province":"上海","address_city":"浦東新區","address_district":null,"address_street":"上海市浦東新區廣蘭路1弄2號345室","price":30.0,"carriage":6.0,"state":"canceled","created_at":"2013-10-24T09:09:28.000Z","payed_at":null,"goods":["營養早餐：火腿麥滿分"],"position":[121.53,31.22],"weight":70.0,"height":172.0,"sex_type":"female","birthday":"1988-01-01"}
[root@es-bulk tmp]# curl -XPOST 'localhost:9200/stuff_orders/_bulk?pretty' --data-binary @test.json
{
  "error" : {
    "root_cause" : [ {
      "type" : "action_request_validation_exception",
      "reason" : "Validation Failed: 1: no requests added;"
    } ],
    "type" : "action_request_validation_exception",
    "reason" : "Validation Failed: 1: no requests added;"
  },
  "status" : 400
}
[root@es-bulk tmp]# curl localhost:9200/stuff_orders/order_list/903713?pretty
{
  "_index" : "stuff_orders",
  "_type" : "order_list",
  "_id" : "903713",
  "found" : false
}
[root@es-bulk tmp]#

產生了報錯，並且資料也的確沒有加成功，原因是在校驗操作請求(action_and_meta_data)時，由於不符合規範，所以報異常

正確匯入方法

解決辦法是將格式糾正過來，加上換行

[root@es-bulk tmp]# vim test.json 
[root@es-bulk tmp]# cat test.json 
{"index":{"_index":"stuff_orders","_type":"order_list","_id":903713}}
{"real_name":"劉備","user_id":48430,"address_province":"上海","address_city":"浦東新區","address_district":null,"address_street":"上海市浦東新區廣蘭路1弄2號345室","price":30.0,"carriage":6.0,"state":"canceled","created_at":"2013-10-24T09:09:28.000Z","payed_at":null,"goods":["營養早餐：火腿麥滿分"],"position":[121.53,31.22],"weight":70.0,"height":172.0,"sex_type":"female","birthday":"1988-01-01"}
[root@es-bulk tmp]# curl -XPOST 'localhost:9200/stuff_orders/_bulk?pretty' --data-binary @test.json
{
  "took" : 36,
  "errors" : false,
  "items" : [ {
    "index" : {
      "_index" : "stuff_orders",
      "_type" : "order_list",
      "_id" : "903713",
      "_version" : 1,
      "_shards" : {
        "total" : 2,
        "successful" : 1,
        "failed" : 0
      },
      "status" : 201
    }
  } ]
}
[root@es-bulk tmp]# curl localhost:9200/stuff_orders/order_list/903713?pretty
{
  "_index" : "stuff_orders",
  "_type" : "order_list",
  "_id" : "903713",
  "_version" : 1,
  "found" : true,
  "_source":{"real_name":"劉備","user_id":48430,"address_province":"上海","address_city":"浦東新區","address_district":null,"address_street":"上海市浦東新區廣蘭路1弄2號345室","price":30.0,"carriage":6.0,"state":"canceled","created_at":"2013-10-24T09:09:28.000Z","payed_at":null,"goods":["營養早餐：火腿麥滿分"],"position":[121.53,31.22],"weight":70.0,"height":172.0,"sex_type":"female","birthday":"1988-01-01"}
}
[root@es-bulk tmp]#

Tip: 當資料量極大時，這樣一個個改肯定不方便，這時可以使用sed指令碼，能很方便的進行批量修改

[[email protected] summary]# sed -ir  's/[}][}][{]/\}\}\n\{/' jjjj.json 
[[email protected] summary]# less jjjj.json

其實就是匹配到合適的地方加上一個換行

記憶體不足

基本上只要遵循前面的操作方式，理想情況下都會很順利地將資料匯入ES，但是實現環境中，總會有各種意外，我就遇到了其中一種：記憶體不足

[root@es-bulk tmp]# time curl -XPOST 'localhost:9200/stuff_orders/_bulk?pretty' --data-binary @es_data.json > /dev/null
  % Total    % Received % Xferd  Average Speed   Time    Time     Time  Current
                                 Dload  Upload   Total   Spent    Left  Speed
 38  265M    0     0   38  102M      0  43.8M  0:00:06  0:00:02  0:00:04 43.9M
curl: (56) Failure when receiving data from the peer

real	0m5.351s
user	0m0.161s
sys	0m0.919s
[root@es-bulk tmp]#

當時百思不得其解，已經反覆確認了資料格式無誤，並且隨機選取其中一些進行匯入測試也沒發現問題，但只要整體一導就出問題，而且每次都一樣

[root@es-bulk tmp]# free -m 
             total       used       free     shared    buffers     cached
Mem:          3949       3548        400          0          1        196
-/+ buffers/cache:       3349        599
Swap:         3951        237       3714
[root@es-bulk tmp]#

系統記憶體明明還有多餘，但是再看到JAVA記憶體時，就隱約感覺到了原因

[root@es-bulk tmp]# ps faux | grep elas
root     14479  0.0  0.0 103252   816 pts/1    S+   16:05   0:00          \_ grep elas
495      19045  0.2 25.6 3646816 1036220 ?     Sl   Mar07  25:45 /usr/bin/java -Xms256m -Xmx1g -Djava.awt.headless=true -XX:+UseParNewGC -XX:+UseConcMarkSweepGC -XX:CMSInitiatingOccupancyFraction=75 -XX:+UseCMSInitiatingOccupancyOnly -XX:+HeapDumpOnOutOfMemoryError -XX:+DisableExplicitGC -Dfile.encoding=UTF-8 -Djna.nosys=true -Des.path.home=/usr/share/elasticsearch -cp /usr/share/elasticsearch/lib/elasticsearch-2.1.1.jar:/usr/share/elasticsearch/lib/* org.elasticsearch.bootstrap.Elasticsearch start -p /var/run/elasticsearch/elasticsearch.pid -d -Des.default.path.home=/usr/share/elasticsearch -Des.default.path.logs=/var/log/elasticsearch -Des.default.path.data=/var/lib/elasticsearch -Des.default.path.conf=/etc/elasticsearch
[root@es-bulk tmp]#

ES和lucene是使用的JAVA，JAVA的記憶體分配大小決定了它們的發揮空間，這裡的初始記憶體為 256M ，這也是大多數情況下的預設配置，但是應對當前的實際資料大小 265M 時就不夠了，雖然官方說會盡量減小使用buffer，但實測下來，系統應該會是首先儘量使用記憶體，通過匯入記憶體的方式來起到顯著加速的效果，但是記憶體不夠時，就直接報錯退出了

解決記憶體不足有兩種思路：

  • 1.調整 Xms 和 Xmx 引數，使其適應業務需求，然後重啟服務使之生效
  • 2.將原來的資料切小，分批匯入

第一種方式，要求停應用和業務，在某些情況下是不具備條件的(得統一協調時間視窗)，那麼就嘗試使用第二種方式,好在text文件的切分也可以使用sed快速完成

[root@es-bulk tmp]# sed  -rn '1,250000p' es_data.json  > es_data1.json
[root@es-bulk tmp]# sed  -rn '250001,500000p' es_data.json  > es_data2.json
[root@es-bulk tmp]# sed  -rn '500001,750000p' es_data.json  > es_data3.json
[root@es-bulk tmp]# sed  -rn '750001,943210p' es_data.json  > es_data4.json
[root@es-bulk tmp]# 
[root@es-bulk tmp]# du -sh es_data*.json
71M	es_data1.json
68M	es_data2.json
71M	es_data3.json
58M	es_data4.json
266M	es_data.json
[root@es-bulk tmp]#
[root@es-bulk tmp]# tail es_data1.json
...
...
[root@es-bulk tmp]# tail es_data2.json
...
...

再依次進行匯入，就發現沒問題了

[root@es-bulk tmp]# time curl -XPOST 'localhost:9200/stuff_orders/_bulk?pretty' --data-binary @es_data1.json > /dev/null
  % Total    % Received % Xferd  Average Speed   Time    Time     Time  Current
                                 Dload  Upload   Total   Spent    Left  Speed
100  101M  100 30.6M  100 70.3M   981k  2253k  0:00:31  0:00:31 --:--:--     0

real	0m33.308s
user	0m0.100s
sys	0m0.390s
[root@es-bulk tmp]#

命令彙總

  • curl -XPOST 'localhost:9200/stuff_orders/_bulk?pretty' --data-binary @test.json
  • curl localhost:9200/stuff_orders/order_list/903713?pretty
  • sed -ir 's/[}][}][{]/\}\}\n\{/' jjjj.json
  • less jjjj.json
  • time curl -XPOST 'localhost:9200/stuff_orders/_bulk?pretty' --data-binary @es_data.json > /dev/null
  • free -m
  • ps faux | grep elas
  • sed -rn '1,250000p' es_data.json > es_data1.json
  • sed -rn '250001,500000p' es_data.json > es_data2.json
  • sed -rn '500001,750000p' es_data.json > es_data3.json
  • sed -rn '750001,943210p' es_data.json > es_data4.json
  • du -sh es_data*.json
  • tail es_data1.json
  • time curl -XPOST 'localhost:9200/stuff_orders/_bulk?pretty' --data-binary @es_data1.json > /dev/null
原文地址 http://soft.dog/2016/03/15/elasticsearch-bulk/

相關推薦

Elasticsearch 批量匯入資料

前言  可以非常方便地進行資料的多維分析，所以大資料分析領域也經常會見到它的身影，生產環境中絕大部分新產生的資料可以通過應用直接匯入，但是歷史或初始資料可能會需要單獨處理，這種情況下可能遇到需要匯入大量資料的情況 這裡簡單分享一下批量匯入資料的操作方法與相關基礎，還有可能

Elasticsearch —— bulk批量匯入資料

在使用Elasticsearch的時候，一定會遇到這種場景——希望批量的匯入資料，而不是一條一條的手動匯入。那麼此時，就一定會需要bulk命令！ 更多內容參考我整理的Elk教程 bulk批量匯入 批量匯入可以合併多個操作，比如index,delete,update,create等等。也可以幫助從一個索引匯

python批量匯入資料進Elasticsearch

ES在之前的部落格已有介紹，提供很多介面，本文介紹如何使用python批量匯入。ES官網上有較多說明文件，仔細研究並結合搜尋引擎應該不難使用。 先給程式碼 #coding=utf-8 from datetime import datetime from elasticse

ElasticSearch中批量匯入資料: _bulk命令

在使用Elasticsearch的時候，一定會遇到這種場景——希望批量的匯入資料，而不是一條一條的手動匯入。那麼此時，就一定會需要bulk命令！ bulk批量匯入 批量匯入可以合併多個操作，比如index,delete,update,create等等。也可以

批量匯入資料（Mysql）報MySQL server has gone away 問題的解決方法

問題分析 首先度娘：mysql出現ERROR : (2006, 'MySQL server has gone away') 的問題意思就是指client和MySQL server之間的連結斷開了。    造成這樣的原因一般是sql操作的時間過長，或者是傳送的資料太大(例如使用inser

使用kettle來根據時間戳或者批次號來批量匯入資料，達到增量的效果。

1、Kettle是一款國外開源的ETL工具，純java編寫，可以在Window、Linux、Unix上執行，資料抽取高效穩定。下載圖形化介面的zip包格式的，直接解壓縮使用即可。安裝部署模式這裡不說了，自己可以根據自己的需求安裝為單機模式或者叢集模式。    Kettle的社群官

Solr批量匯入資料

1、從solr 4.10.3 的dist 下 所需Jar包複製一份到 solr_home的索引庫（db1 是我自己新建的索引庫）下lib下 如果沒有就新建 2、在索引庫（db1）的核心配置檔案（solrconfig.xml）中 新增請求處理器。 <requestHand

將根據時間戳增量資料方案修改為根據批次號增量資料方案 使用kettle來根據時間戳或者批次號來批量匯入資料，達到增量的效果。

1、之前寫過根據時間戳來增量資料，時間戳增量資料存在一定的缺點，就是如果開啟自動的話，以後如果因為某個外在因素出錯了，那麼這個開始時間和結束時間不好控制，那麼就可能造成一些其他資料量不準的情況，但是根據批次號不會出現這個問題： 使用kettle來根據時間戳或者批次號來批量匯入資料，達到增量的效果。

react下批量匯入資料無法覆蓋原資料中的input值的問題（利用antd design外掛）

最近在寫一個需求，要求批量匯入資料後要把原來的列表資料全部覆蓋，包括匯入的資料與原來的列表資料重複的值。 如果匯入的資料中與原來列表的資料沒有重複的值，那就重新setState一下就好；那如果列表中有資料A，並且A的一個屬性通過input編輯過，此時再批量匯入資料包含資料A的話，那就不能覆蓋已經

C#解析Mysql的sql指令碼實現批量匯入資料

最近老大對我們做的資料匯入功能意見挺大，資料量一上來，匯入時間就很長，嚴重影響使用者體驗。因此，不得不花時間搗鼓了一下資料匯入的效能優化問題 原始程式碼： MySqlCommand command = new MySqlCommand(); command.Connection = conn

Ne4j如何使用python批量匯入資料

說到neo4j的批量匯入資料，我想你一定會想到如下幾點： import tools load file neo4j driver for python/java… 前兩種必須要資料檔案存在檔案系統才可以執行。 但是如果你的資料是以流資料的形式持續獲取的呢

Java實現Excel批量匯入資料

Excel的批量匯入是很常見的功能，這裡採用 Jxl實現，資料量或樣式要求較高可以採用 poi 框架環境：Spring + SpringMvc（註解實現） 首先匯入依賴jar包 <dependency> <groupId>ne

python實現Phoenix批量匯入資料

官網文件： Phoenix provides two methods for bulk loading data into Phoenix tables: Single-threaded client loading tool for CSV formatted data

HBase利用bulk load批量匯入資料

OneCoder只是一個初學者，記錄的只是自己的一個過程。不足之處還望指導。 看網上說匯入大量資料，用bulk load的方式效率比較高。bulk load可以將固定格式的資料檔案轉換為HFile檔案匯入，當然也可以直接匯入HFile檔案。所以

EXCEL批量匯入資料MVC

using NPOI.HSSF.UserModel; //批量匯入的頁面Action public ActionResult ExcelAppendData() { return View(); } /// /// 匯入Excel方法 /// 作者

用java實現從txt文字檔案批量匯入資料至資料庫

今天同事讓我準備一個專案的測試資料，要向一個表中插入上千條記錄，並且保證每條記錄內容不同，如果用手工一條一條插入肯定是不可能，也不會有哪個SB去做這樣的事，我最開始想到了用迴圈，但要求插入的記錄內容不能相同，用迴圈實現比較麻煩，於是我想到了將記錄從文字檔案匯入至資料庫（其實

簡單三層使用DataTable向資料庫表批量匯入資料---向SqlServer一張表中匯入資料

向資料庫的一張表中新增資料，可以採用單個新增，即一條資料、一條資料的新增；也可以採用批量匯入，依次將好些條資料寫入資料庫的一張表中。文字藉助例項《新增系列資訊》講解一種向資料庫批量匯入資料的方法。 1.介面設計 觀看一下新增系列資訊部分的介面設計： 本

Hbase批量匯入資料，支援多執行緒同時操作

/** * HBase操作工具類：快取模式多執行緒批量提交作業到hbase * * @Auther: ning.zhang * @Email: [email protected] * @CreateDate: 2018/7/30 */ public c

31.Elasticsearch批量匯入本地Json檔案Java實現(ES檔案同步)

題記產品開發需要，我們需要將網際網路採集的資料儲存到ES中，以實現資料的全文檢索。網際網路採集的資料，往往格式雜亂，需要先進行資料清洗操作。而ES支援的入庫格式，json格式資料會相對方便些。本文主要介紹，如何將格式化的Json檔案批量插入到ES中。1、需提前做的工作1）設計

Sqlserver資料庫批量匯入資料

最近抽空整理了一下專案中運用到的公共方法，打算增加自己的DLL檔案的內容。突然發現sqlserver資料批量插入的一個東西。好像我的DLL檔案裡面沒有，所以就加上了。這裡說明一下：專案中所有的公共檔案都是我自己建的，公共類庫中的大部分方法都是我自己寫的，不存在什麼洩露程式碼