數據怎麼導入到es當中的

A. 電腦怎麼用es文件管理器傳文件

1、首先需要打開ES文件管理器，界面直觀，文件自動分類顯示。

B. java導數據到esid重復

，代碼主要邏輯如下：

// 讀取要導入數據的文件
BufferedReader br = new BufferedReader(new FileReader(
"D:\\test\\test.txt"));
String json = null;
int count = 0;
// 開啟批量插入
BulkRequestBuilder bulkRequest = client.prepareBulk();
while ((json = br.readLine()) != null) {
bulkRequest.add(client.prepareIndex("test", "all")
.setSource(json));
// 每一千條提交一次
if (count % 1000 == 0) {
bulkRequest.execute().actionGet();
System.out.println("提交了：" + count);
}
count++;
}
bulkRequest.execute().actionGet();
System.out.println("插入完畢");
br.close();
登錄後復制

運行後發現一個問題，我100多萬條的數據，導入到es中怎麼生成了1000多萬條，而且還是在沒有完全導入的情況下
然後用小批量數據導入到es,再把這些數據導出來，發現有好多重復的數據

為什麼會重復呢，原因是在每一千條提交一次代碼這塊，第一次一千條提交了，並沒有把bulkRequest置空，所以第二次提交的時候，會提交兩千條，包括第一次已經提交的一千條，然後我們自己也沒有設置_id,所以es會自動給數據生成一個_id,即使是重復的數據，搞清楚了原因，下面來說解決方法，主要有兩種：

第一種就是在提交了一千條後，對bulkRequest進行重置，因為bulkRequest並沒有重置的方法，所以可以新建一個bulkRequest，類似於重置，具體代碼如下：

// 讀取要導入數據的文件
BufferedReader br = new BufferedReader(new FileReader(
"D:\\test\\test.txt"));
String json = null;
int count = 0;
// 開啟批量插入
BulkRequestBuilder bulkRequest = client.prepareBulk();
while ((json = br.readLine()) != null) {
bulkRequest.add(client.prepareIndex("test", "all")
.setSource(json));
// 每一千條提交一次
if (count % 1000 == 0) {
bulkRequest.execute().actionGet();
//此處新建一個bulkRequest，類似於重置效果
bulkRequest = client.prepareBulk();
System.out.println("提交了：" + count);
}
count++;
}
bulkRequest.execute().actionGet();
System.out.println("插入完畢");
br.close();
登錄後復制

第二種就是自己設置_id,確保每一條數據只有一個_id,這樣的話，即使數據重復了，因為_id是一樣的，所以es會進行更新，這樣的話並沒有從根源上解決數據重復的問題，只是重復數據會更新，這樣的話效率會慢，具體代碼如下：
// 讀取要導入數據的文件
BufferedReader br = new BufferedReader(new FileReader(
"D:\\test\\test.txt"));
String json = null;
int count = 0;
// 開啟批量插入
BulkRequestBuilder bulkRequest = client.prepareBulk();
while ((json = br.readLine()) != null) {
//設置_id為count
bulkRequest.add(client.prepareIndex("test", "all",
String.valueOf(count)).setSource(json));
// 每一千條提交一次
if (count % 1000 == 0) {
bulkRequest.execute().actionGet();
//此處新建一個bulkRequest，類似於重置效果
System.out.println("提交了：" + count);
}
count++;
}
bulkRequest.execute().actionGet();
System.out.println("插入完畢");
br.close();
登錄後復制

建議使用第一種方法，效率會快很多。

C. spark->es快速導入數據

elasticsearch-spark 提供了saveToEs api以支持快速導入數據。但es集群線程池有限，在大量寫入數據的同時，對cpu的壓力非常大，影響線上es的查詢服務。如果能參考hbase 的bulkload方法，對es也採用「bulkload」模式，寫入性能會有巨大提升。核心思想是通過spark作業生成es的lucene文件，並通過網路傳輸，寫入es的數據文件。

本方案參考滴滴的fastIndex： 滴滴FastIndex
採用spark改寫，部分特性適應了公司的原始流程，會有不一樣的地方。如您採用的是spark saveToEs需要通過該方法進行改寫，可參考。

git地址為： https://github.com/Dengyu123/fast-es-rdd

D. 記事本的數據怎麼導入到excel中

1、使用記事本編輯數據之後，我們導入到excel中。

E. 【elasticsearch實戰】mysql的數據如何遷移到es中

如果你被上述問題困擾過，可以參考以下方案

這里需要介紹三種欄位的type，分別是 object 、 nested 、 join

現在有個問題，下面的數據如何存入到es中呢，它對應的mapping應該是什麼樣的呢

name、url這些欄位好處理，直接設定欄位 "type" : "text" 或者 "type" : "keword" 或者

就行了，但是對於address和links，這種裡麵包含json對象，或者數組的，怎麼處理呢。這里可以採用 "type" : "object" 來處理。如下

可能會對links有疑問，它明明是數組，卻怎麼和address的設置類似。其實es中是沒有單獨的數組這一類型，因為他所有的欄位都支持數組，比如你是text，你可以放多個值進去，以name為例，你可以放 "name":["張三", "李四"] 這樣的數據進去。
而且，es默認對這種嵌套結構建立的索引就是object類型， "type": "object" 可以省略 😂
於是可以變為下面這樣

甚至，通過添加properties，可以無限嵌套下去。

下面說object類型的缺點了，缺點也是由它本身結構導致的

對於數組結構，是這么存儲數據的,以上面的address為例,他會把json結構平鋪開，然後把所有這個欄位的值放在平鋪後的欄位上：

這在查詢時就出現問題了，本來Google和 http://www.google.com 是綁定的，但是這種結構無法滿足這種綁定的關系，也就是如果你想查name是Bai，並且url是 http://www.google.com 的，竟然也能查出來😂，而這和前面所插入的文檔內容不符。

所以需要nested結構和join結構出場了

嵌套結構解決了我們查詢嵌套文檔欄位的問題，同樣的，也可以解決，在es中實現類似mysql的join查詢的問題。
外鍵就需要設置為nested（雖然現在設計表幾乎不用外鍵約束了，但外鍵的邏輯還是在的 😂 )
另外，nested欄位本身會形成一個文檔，只不過是嵌套在大的文檔下，所以在統計索引的文檔數時，實際上是最外層的文檔數加上nested欄位形成的文檔數

這里需要注意，以nested裡面的欄位為查詢條件，需要修改下查詢DSL，在外層加一層nested，每有一層nested嵌套關系，就需要加一層

由於es本身對文檔通過nested欄位進行了綁定，索引更新數據時，整個文檔都會被替換，代價會大一些，但是由於關系綁定好了，查詢會快一些。這里的代價大一些，查詢快一些自然就是和join類型對比啦。

join 其實有父子文檔的概念，父文檔通過一個欄位關聯一個子文檔，
這個結構比較復雜的是在你推數據時，需要指定對應的父文檔是哪個
mapping結構如下

解釋一下

優點就是更新數據時，不用連帶著父子文檔一起改，缺點是查詢效率不如nested結構

以後再說吧😂

F. ES跨集群數據遷移

reindex是Elasticsearch提供的一個api介面，可以把數據從源ES集群導入到當前的ES集群，同樣實現了數據的遷移

如果是A集群 --> B集群，就需要在B中的elasticsearch.yml 設置A地址為白名單

在目標集群（B集群 116）的elasticsearch.yml配置文件，設置遠程集群的白名單，添加如下配置：

對mapping有要求的，提前創建好索引，再執行數據遷移。

1、get /索引名稱 獲取索引
2、put /索引名稱 + mapping信息。創建索引

這個帖子寫的很好，很好
https://blog.csdn.net/qq_21383435/article/details/108953326

G. 使用canal將mysql同步到es中

因為自己項目中需要用到mysql數據同步到es中，查找了相關資料最後決定用canal來做，所以便有了本文，下面一起來看如何使用canal吧

根據 https://github.com/alibaba/canal 上的原理解釋，我們知道 canal 會模擬 mysql slave 的交互協議，偽裝自己為 mysql slave，然後向 mysql master 發送 mp 協議。

mysql master 收到 mp 請求，開始推送 binary log 給 slave(也就是 canal)，然後 canal 解析 binary log 對象(原始為 byte流)。

經 canal 解析過的對象，我們使用起來就非常的方便了。

再根據 https://github.com/alibaba/canal/releases 提供的版本信息，你會發現 canal 其實相當於一個中間件，專門用來解析 MySQL 的 binlog 日誌。canal 解析好了之後，會封裝成一個數據對象，通過 protobuf3.0 協議進行交互，讓 canal 客戶端進行消費。

根據上面的解釋，以及 canal 提供的版本信息，我們在使用 canal 的時候，首選要安裝一個 canal.deployer-1.1.4.tar.gz 進行解析 MySQL 的 binlog 日誌。

下載後，復制 canal.deployer-1.1.4.tar.gz 到 MySQL 主機上，比如放在 /usr/local/soft/目錄下。然後依次執行下面的命令：

然後修改 canal 的配置文件 vim conf/example/instance.properties

這三項改成你自己的，比如我的配置如下：

然後保存並退出。（VI 模式下，按 Esc 輸入 :wq 回車退出。）

接著，我們檢查一下 MySQL 的配置。確定版本和是否開啟了 binlog 日誌，以及日誌格式。

canal 支持 binlog 格式為 ROW 的模式。如果你沒開啟 binlog，並且格式是非 row 的，建議修改一下 mysql 的配置文件。

執行 mysql –help | grep my.cnf 找到 mysql 的 my.cnf 文件。

執行 vim /etc/my.cnf 命令。添加下面 3 個配置。

然後保存並退出。

接著執行 sudo service mysqld restart 重啟 MySQL。

需要注意的是你的 mysql 用戶，必須要有 REPLICATION SLAVE 許可權。該許可權授予 slave 伺服器以該賬戶連接 master 後可以執行 replicate 操作的權利。

如果沒有許可權，則使用 root 賬戶登錄進 MySQL，執行下面的語句，創建用戶，分配許可權。

MySQL 啟動後，就可以開啟 canal 服務了。

開啟後，觀察 canal 服務的日誌，確保服務正常。

查看 canal 的日誌

確定沒有問題後，開始編寫我們的測試程序。

pom.xml 中導入下面的依賴。

使用JAVA進行測試

然後執行 main 方法。你再修改修改 MySQL 中的數據，你會發現所有改變都同步過來了。上面是使用的Java代碼進行運行，如果想用canal.adapter來進行運行可以下載

放入伺服器中，依次執行下面命令

然後修改配置文件 ：

然後將需要運行存儲到es的的yml文件放入到

目錄下。例如：

然後開啟canal-adapter服務

/usr/local/soft/canal-adapter/bin/startup.sh

查看 canal-adapter 的日誌，確定沒有問題後修改數據 就可以同步到es了

注意：

1、canal-adapter自帶mysql連接使用的5.x的，如果自己安裝的是高版本的mysql需要自己去/usr/local/soft/canal-adapter/lib增加對應的jar包

2、因項目中同步es使用的sql中有資料庫中沒有的欄位，導致原生程序一直報異常，後修改源碼中

加了一個判斷後才可以

3、es中使用的date欄位類型和資料庫中不一致，所以這里又修改了部分源碼兼容我們項目中的類型

可以根據各自情況修改。

H. elasticmp實現es數據導入導出

elasticmp 提供了多種導入導出數據的方式，可以 index <-> index 、 index <-> .json 文件，還支持將 index 查詢結果導出到 .json 文件。執行的命令也很簡單，只需指定數據來源 input 、數據輸出 output 、數據類型 type 即可。

a.導出 index 數據到 .json 文件

b.導入 .json 文件中的數據到 es

在導入數據前可以先把測試用的index數據清空

I. logstash 怎麼將數據導入ES

在配置文件
input {
file {
type => "json"
path => "/home/hadoop/xinwang_XW351464_2110"
}
}
output {
elasticsearch {
cluster => "es_master"
#manage_template => false
embedded => true
index => "huhu"
}
}
版本是logstash-1.3.2，es版本是1.1.1
錯誤提示
java -jar logstash-1.3.2-flatjar.jar agent -f ogstash-syslog.conf
Using milestone 2 input plugin 'file'. This plugin should be stable, but if you see strange behavior, please let us know! For more information on plugin milestones, see http://logstash.net/docs/1.3.2/plugin-milestones {:level=>:warn}
log4j, [2014-10-31T11:55:01.977] WARN: org.elasticsearch.discovery: [Jocasta] waited for 30s and no initial state was set by the discovery
Unable to check template. Automatic template management disabled. {:error=>"waited for [30s]",
Redhat 5.7 64bit / CentOS 5.x
JDK 1.6.0_45
logstash 1.3.2 (內帶kibana)
elasticsearch 0.90.10
redis 2.8.4
對應下版本

J. Logstash導入csv到es

[TOC]

如導入的文件為 phone_area.csv ，文件格式如下：

定義配置文件 phone_area_imp.conf ：

執行導入命令：

導入界面， 將一直停留在命令行，不退出 ：

當csv文件發生變動時，仍然會同步新數據到es中 。