摘要：最近為了導庫的問題，費了一些周折。可定制的數據導入工具基于的主要功能完全使用實現數據從到的遷移可批量導入多張表可自定義的數據遷移的規則數據表字段關系字段過濾使用正則進行異常處理可自定義的異步分片導入方式，數據導入效率更高。

最近為了es導庫的問題，費了一些周折。于是乎做了一個小工具（用過npm的一些jdbc的導庫工具，感覺還不夠好用），這里拋磚引玉，自薦一下下，歡迎同道兄弟吐槽和參與，我會不定時的更新這個擴展。

A customizable importer from mysql to elasticsearch.

可定制的 elasticsearch 數據導入工具 ——基于 elasticsearch 的 JS API

完全使用 JS 實現數據從 MySQL 到 elasticsearch 的遷移；

可批量導入多張 MySQL 表；

可自定義的數據遷移的規則（數據表/字段關系、字段過濾、使用正則進行異常處理）；

可自定義的異步分片導入方式，數據導入效率更高。

npm install mysql_2_elasticsearch

var esMysqlRiver = require("mysql_2_elasticsearch");

var river_config = {
  mysql: {
    host: "127.0.0.1",
    user: "root",
    password: "root",
    database: "users",
    port: 3306
  },
  elasticsearch: {
    host_config: {               // es客戶端的配置參數
      host: "localhost:9200",
      // log: "trace"
    },
    index: "myIndex"
  },
  riverMap: {
    "users => users": {}         // 將數據表 users 導入到 es 類型: /myIndex/users
  }
};


/*
** 以下代碼內容：
** 通過 esMysqlRiver 方法進行數據傳輸，方法的回調參數(一個JSON對象) obj 包含此次數據傳輸的結果
** 其中：
** 1. obj.total    => 需要傳輸的數據表數量
** 2. obj.success  => 傳輸成功的數據表數量
** 3. obj.failed   => 傳輸失敗的數據表數量
** 4. obj.result   => 本次數據傳輸的結論
*/

esMysqlRiver(river_config, function(obj) {
  /* 將傳輸結果打印到終端 */
  console.log("
---------------------------------");
  console.log("總傳送：" + obj.total + "項");
  console.log("成功：" + obj.success + "項");
  console.log("失敗：" + obj.failed + "項");
  if (obj.result == "success") {
    console.log("
結論：全部數據傳送完成！");
  } else {
    console.log("
結論：傳送未成功...");
  }
  console.log("---------------------------------");
  /* 將傳輸結果打印到終端 */
});

var esMysqlRiver = require("mysql_2_elasticsearch");

/*
** mysql_2_elasticsearch 的相關參數配置(詳情見注釋)
*/

var river_config = {

  /* [必需] MySQL數據庫的相關參數(根據實際情況進行修改) */
  mysql: {
    host: "127.0.0.1",
    user: "root",
    password: "root",
    database: "users",
    port: 3306
  },

  /* [必需] es 相關參數(根據實際情況進行修改) */
  elasticsearch: {
    host_config: {               // [必需] host_config 即 es客戶端的配置參數，詳細配置參考 es官方文檔
      host: "localhost:9200",
      log: "trace",
      // Other options...
    },
    index: "myIndex",            // [必需] es 索引名
    chunkSize: 8000,             // [非必需] 單分片最大數據量，默認為 5000 (條數據)
    timeout: "2m"                // [非必需] 單次分片請求的超時時間，默認為 1m
    //(注意：此 timeout 并非es客戶端請求的timeout，后者請在 host_config 中設置)
  },

  /* [必需] 數據傳送的規則 */
  riverMap: {
    "users => users": {            // [必需] "a => b" 表示將 mysql數據庫中名為 "a" 的 table 的所有數據 輸送到 es中名為 "b" 的 type 中去
      filter_out: [                // [非必需] 需要過濾的字段名，即 filter_out 中的設置的所有字段將不會被導入 elasticsearch 的數據中
        "password",
        "age"
      ],
      exception_handler: {           // [非必需] 異常處理器，使用JS正則表達式處理異常數據，避免 es 入庫時由于類型不合法造成數據缺失
        "birthday": [                // [示例] 對 users 表的 birthday 字段的異常數據進行處理
          {
            match: /NaN/gi,          // [示例] 正則條件(此例匹配字段值為 "NaN" 的情況)
            writeAs: null            // [示例] 將 "NaN" 重寫為 null
          },
          {
            match: /(d{4})年/gi,    // [示例] 正則表達式(此例匹配字段值為形如 "2016年" 的情況)
            writeAs: "$1.1"          // [示例] 將 "2015年" 樣式的數據重寫為 "2016.1" 樣式的數據
          }
        ]
      }
    },
    // Other fields" options...
  }

};


/*
** 將傳輸結果打印到終端
*/

esMysqlRiver(river_config, function(obj) {
  console.log("
---------------------------------");
  console.log("總傳送：" + obj.total + "項");
  console.log("成功：" + obj.success + "項");
  console.log("失敗：" + obj.failed + "項");
  if (obj.result == "success") {
    console.log("
結論：全部數據傳送完成！");
  } else {
    console.log("
結論：傳送未成功...");
  }
  console.log("---------------------------------");
});

elasticsearch數據導入前請先配置好數據的 mapping；

host_config 更多參數設置詳見 es官方API文檔；

mysql 表的自增 id 自動替換為 表名+_id 的格式，如：users_id；

如出現數據缺失情況，請注意查看 elasticsearch 終端進程或日志，找出未成功導入的數據，通過設置 exception_handler 參數處理它。