如何在Ubuntu14.04上使用Transporter将转换后的数据从MongoDB同步到Elasticsearch
介绍
Elasticsearch 有助于对数据进行全文搜索，而 MongoDB 擅长存储数据。 使用 MongoDB 存储数据并使用 Elasticsearch 进行搜索是一种常见的架构。
很多时候，您可能会发现需要将数据从 MongoDB 批量迁移到 Elasticsearch。 为此编写自己的程序，虽然是一个很好的练习，但可能是一项乏味的任务。 有一个名为 Transporter 的出色开源实用程序，由 Compose（数据库云平台）开发，可以非常高效地完成这项任务。
本教程向您展示如何使用开源实用程序 Transporter 通过自定义转换将数据从 MongoDB 快速复制到 Elasticsearch。
目标
在本文中，我们将介绍如何使用 Transporter 实用程序在 Ubuntu 14.04 上将数据从 MongoDB 复制到 Elasticsearch。
我们将从一个快速概述开始，向您展示如何安装 MongoDB 和 Elasticsearch，尽管我们不会详细介绍这两个系统中的数据建模。 如果您已经安装了这两个步骤，请随意快速浏览安装步骤。
然后我们将继续讨论Transporter。
其他版本的 Ubuntu 以及其他 Linux 发行版的说明类似。
先决条件
请完成以下先决条件。
- Ubuntu 14.04 液滴
- sudo 用户
第 1 步 — 安装 MongoDB
导入 MongoDB 存储库的公钥。
sudo apt-key adv --keyserver hkp://keyserver.ubuntu.com:80 --recv 7F0CEB10
为 MongoDB 创建一个列表文件。
echo 'deb http://downloads-distro.mongodb.org/repo/ubuntu-upstart dist 10gen' | sudo tee /etc/apt/sources.list.d/mongodb.list
重新加载本地包数据库。
sudo apt-get update
安装 MongoDB 包：
sudo apt-get install -y mongodb-org
请注意，每个包都包含相关的版本号。
安装完成后，您可以启动、停止和检查服务的状态。 安装后会自动启动。
尝试连接到作为服务运行的 MongoDB 实例：
mongo
如果它已启动并运行，您将看到如下内容：
MongoDB shell version: 2.6.9
connecting to: test
Welcome to the MongoDB shell.
For interactive help, type "help".
For more comprehensive documentation, see
http://docs.mongodb.org/
Questions? Try the support group
http://groups.google.com/group/mongodb-user
这意味着数据库服务器正在运行！ 您现在可以退出：
exit
第 2 步 — 安装 Java
Java 是 Elasticsearch 的先决条件。 让我们现在安装它。
首先，添加存储库：
sudo apt-add-repository ppa:webupd8team/java
再次更新您的包裹清单：
sudo apt-get update
安装 Java：
sudo apt-get install oracle-java8-installer
当提示接受许可时，选择 <Ok>，然后选择 <Yes>。
第 3 步 — 安装 Elasticsearch
现在我们将安装 Elasticsearch。
首先，创建一个新目录，您将在其中安装搜索软件，然后进入该目录。
mkdir ~/utils cd ~/utils
访问 Elasticsearch 的 下载页面 以查看最新版本。
现在下载最新版本的 Elasticsearch。 在撰写本文时，最新版本是 1.5.0。
wget https://download.elasticsearch.org/elasticsearch/elasticsearch/elasticsearch-1.5.0.zip
安装解压：
sudo apt-get install unzip
解压存档：
unzip elasticsearch-1.5.0.zip
导航到您提取它的目录：
cd elasticsearch-1.5.0
通过发出以下命令启动 Elasticsearch：
bin/elasticsearch
Elasticsearch 需要几秒钟才能启动。 您会看到一些启动日志。 Elasticsearch 现在将在终端窗口中运行。
注意： 在某些时候您可能希望将 Elasticsearch 作为服务运行，以便您可以使用
sudo service elasticsearch restart和类似命令来控制它； 有关提示，请参阅有关 Upstart 的教程。 或者，您可以从 Ubuntu 的存储库安装 Elasticsearch，尽管您可能会获得旧版本。
保持此终端打开。 在 另一个终端窗口 中与您的服务器建立另一个 SSH 连接，并检查您的实例是否已启动并正在运行：
curl -XGET http://localhost:9200
9200 是 Elasticsearch 的默认端口。 如果一切顺利，您将看到类似于下图的输出：
{
"status" : 200,
"name" : "Northstar",
"cluster_name" : "elasticsearch",
"version" : {
"number" : "1.5.0",
"build_hash" : "927caff6f05403e936c20bf4529f144f0c89fd8c",
"build_timestamp" : "2015-03-23T14:30:58Z",
"build_snapshot" : false,
"lucene_version" : "4.10.4"
},
"tagline" : "You Know, for Search"
}
注意：对于本文的后面部分，当您将复制数据时，请确保 Elasticsearch 正在运行（并且在端口 9200 上）。
第 4 步 — 安装 Mercurial
接下来我们将安装修订控制工具 Mercurial。
sudo apt-get install mercurial
验证 Mercurial 是否已正确安装：
hg
如果安装正确，您将获得以下输出：
Mercurial Distributed SCM basic commands: . . .
第 5 步 — 安装 Go
Transporter 是用 Go 语言编写的。 因此，您需要在系统上安装 golang。
sudo apt-get install golang
为了让 Go 正常工作，您需要设置以下环境变量：
从 $HOME 目录为 Go 创建一个文件夹：
mkdir ~/go; echo "export GOPATH=$HOME/go" >> ~/.bashrc
更新您的路径：
echo "export PATH=$PATH:$HOME/go/bin:/usr/local/go/bin" >> ~/.bashrc
退出您当前的 SSH 会话并重新登录。 您可以只关闭您一直在工作的会话并保持 Elasticsearch 会话运行。 此步骤对于更新环境变量至关重要。 再次登录，并验证您的变量是否已添加：
echo $GOPATH
这应该会显示 Go 的新路径。 在我们的例子中，它将是：
/home/sammy/go
如果它没有正确显示路径，请仔细检查本节中的步骤。
一旦我们的 $GOPATH 设置正确，我们需要通过构建一个简单的程序来检查 Go 是否安装正确。
创建一个名为 hello.go 的文件并将以下程序放入其中。 您可以使用任何您想要的文本编辑器。 我们将在本文中使用 nano 文本编辑器。 键入以下命令以创建新文件：
nano ~/hello.go
现在将下面这个简短的“Hello, world”程序复制到新打开的文件中。 该文件的全部目的是帮助我们验证 Go 是否正常工作。
package main;
import "fmt"
func main() {
fmt.Printf("Hello, world\n")
}
完成后，按 CTRL+X 退出文件。 它将提示您保存文件。 按 Y，然后按 ENTER。 它会询问您是否要更改文件名。 再次按 ENTER 保存当前文件。
然后，从您的主目录中，使用 Go 运行该文件：
go run hello.go
你应该看到这个输出：
Hello, world
如果您看到“Hello, world”消息，则说明 Go 已正确安装。
现在进入 $GOPATH 目录并创建子目录 src、pkg 和 bin。 这些目录构成了 Go 的工作空间。
cd $GOPATH mkdir src pkg bin
src包含组织成包的 Go 源文件（每个目录一个包）pkg包含包对象bin包含可执行命令
第 6 步 — 安装 Git
我们将使用 Git 来安装 Transporter。 使用以下命令安装 Git：
sudo apt-get install git
第 7 步 — 安装 Transporter
现在创建并移动到 Transporter 的新目录。 由于该实用程序是由 Compose 开发的，因此我们将目录称为 compose。
mkdir -p $GOPATH/src/github.com/compose cd $GOPATH/src/github.com/compose
这是 compose/transporter 将被安装的地方。
克隆 Transporter GitHub 存储库：
git clone https://github.com/compose/transporter.git
进入新目录：
cd transporter
取得 /usr/lib/go 目录的所有权：
sudo chown -R $USER /usr/lib/go
确保为 GCC 安装了 build-essential：
sudo apt-get install build-essential
运行 go get 命令获取所有依赖项：
go get -a ./cmd/...
此步骤可能需要一段时间，因此请耐心等待。 完成后，您可以构建 Transporter。
go build -a ./cmd/...
如果一切顺利，它将在没有任何错误或警告的情况下完成。 通过运行以下命令检查是否正确安装了 Transporter：
transporter
你应该看到这样的输出：
usage: transporter [--version] [--help] <command> [<args>]
Available commands are:
about Show information about database adaptors
eval Eval javascript to build and run a transporter application
. . .
这样安装就完成了。 现在，我们需要将 MongoDB 中的一些测试数据同步到 Elasticsearch。
故障排除：
如果您收到以下错误：
transporter: command not found
这意味着您的 $GOPATH 未添加到您的 PATH 变量中。 检查您是否正确执行了命令：
echo "export PATH=$PATH:$HOME/go/bin:/usr/local/go/bin" >> ~/.bashrc
尝试注销并重新登录。 如果错误仍然存在，请改用以下命令：
$GOPATH/bin/transporter
第 8 步 - 创建示例数据
现在我们已经安装了所有东西，我们可以继续进行数据同步部分。
连接到 MongoDB：
mongo
您现在应该看到 MongoDB 提示符 >。 创建一个名为 foo 的数据库。
use foo
将一些示例文档插入名为 bar 的集合中：
db.bar.save({"firstName": "Robert", "lastName": "Baratheon"});
db.bar.save({"firstName": "John", "lastName": "Snow"});
选择刚才输入的内容：
db.bar.find().pretty();
这应该会显示如下所示的结果（ObjectId 在您的机器上会有所不同）：
{
"_id" : ObjectId("549c3ef5a0152464dde10bc4"),
"firstName" : "Robert",
"lastName" : "Baratheon"
}
{
"_id" : ObjectId("549c3f03a0152464dde10bc5"),
"firstName" : "John",
"lastName" : "Snow"
}
现在您可以退出数据库：
exit
一些术语：
- MongoDB 中的 database 类似于 Elasticsearch 中的 index
- MongoDB 中的 collection 类似于 Elasticsearch 中的 type
我们的最终目标是将来自 foo 数据库的 bar 集合的数据从 MongoDB 同步到 foo 索引的 bar 类型在弹性搜索中。
第 9 步 — 配置 Transporter
现在，我们可以继续进行配置更改，将数据从 MongoDB 迁移到 Elasticsearch。 Transporter 需要一个配置文件 (config.yaml)、一个转换文件 (myTransformation.js) 和一个应用程序文件 (application.js)
- 配置文件指定节点、类型和 URI
- 应用程序文件指定从源到目标的数据流以及可选的转换步骤
- 转换文件将转换应用于数据
注意： 本节中的所有命令都假定您正在执行传输器目录中的命令。
移动到 transporter 目录：
cd ~/go/src/github.com/compose/transporter
配置文件
如果您愿意，可以查看示例 config.yaml 文件。 我们将备份原始内容，然后将其替换为我们自己的内容。
mv test/config.yaml test/config.yaml.00
新文件类似，但更新了一些 URI 和一些其他设置以匹配我们服务器上的内容。 让我们从这里复制内容并粘贴到新的 config.yaml 文件中。 再次使用 nano 编辑器。
nano test/config.yaml
将以下内容复制到文件中。 完成后，如前所述保存文件。
# api:
# interval: 60s
# uri: "http://requestb.in/13gerls1"
# key: "48593282-b38d-4bf5-af58-f7327271e73d"
# pid: "something-static"
nodes:
localmongo:
type: mongo
uri: mongodb://localhost/foo
es:
type: elasticsearch
uri: http://localhost:9200/
timeseries:
type: influx
uri: influxdb://root:root@localhost:8086/compose
debug:
type: file
uri: stdout://
foofile:
type: file
uri: file:///tmp/foo
注意 nodes 部分。 与原始文件相比，我们稍微调整了 localmongo 和 es 节点。 Nodes 是各种数据源和目的地。 Type 定义节点的类型。 例如，
mongo表示它是一个 MongoDB 实例/集群elasticsearch表示它是一个 Elasticsearch 节点file表示它是一个纯文本文件
uri 将给 API 端点以与节点连接。 如果未指定，默认端口将用于 MongoDB (27017)。 由于我们需要从 MongoDB 的 foo 数据库中捕获数据，因此 URI 应如下所示：
mongodb://localhost/foo
同样，Elasticsearch 的 URI 将如下所示：
http://localhost:9200/
保存 config.yaml 文件。 您无需进行任何其他更改。
申请文件
现在，打开test目录下的application.js文件。
nano test/application.js
将文件的示例内容替换为如下所示的内容：
Source({name:"localmongo", namespace:"foo.bar"})
.transform({filename: "transformers/addFullName.js"})
.save({name:"es", namespace:"foo.bar"});
保存文件并退出。 这是我们管道的简要说明。
Source(options)标识从中获取数据的源transform指定对每条记录应用什么转换save(options)标识保存数据的位置
选项包括：
name:出现在config.yaml文件中的节点名称namespace:标识数据库和表名； 它必须由点限定（ . ）
转换文件
现在，拼图的最后一块是转型。 如果你还记得，我们在 MongoDB 中存储了两条记录，分别是 firstName 和 lastName。 当您将数据从 MongoDB 同步到 Elasticsearch 时，您可以在这里看到转换数据的真正威力。
假设我们希望存储在 Elasticsearch 中的文档具有另一个名为 fullName 的字段。 为此，我们需要创建一个新的转换文件 test/transformers/addFullName.js。
nano test/transformers/addFullName.js
将以下内容粘贴到文件中。 如前所述保存并退出。
module.exports = function(doc) {
doc._id = doc._id['$oid'];
doc["fullName"] = doc["firstName"] + " " + doc["lastName"];
return doc
}
第一行对于处理 Transporter 处理 MongoDB 的 ObjectId() 字段的方式是必要的。 第二行告诉 Transporter 连接 firstName 和 lastName 以形成 fullName。
这是示例的简单转换，但使用一点 JavaScript，您可以在准备搜索数据时进行更复杂的数据操作。
第 10 步 — 执行转换
现在我们已经完成了设置，是时候同步和转换我们的数据了。
确保 Elasticsearch 正在运行！ 如果没有，请在 新终端 窗口中重新启动它：
~/utils/elasticsearch-1.5.0/bin/elasticsearch
在您的 原始终端 中，确保您在 transporter 目录中：
cd ~/go/src/github.com/compose/transporter
执行以下命令复制数据：
transporter run --config ./test/config.yaml ./test/application.js
Transporter 的 run 命令需要两个参数。 第一个是配置文件，第二个是应用程序文件。 如果一切顺利，该命令将完成且没有任何错误。
通过我们的转换检查 Elasticsearch 以验证数据是否被复制：
curl -XGET localhost:9200/foo/bar/_search?pretty=true
你会得到这样的结果：
{
"took" : 10,
"timed_out" : false,
"_shards" : {
"total" : 5,
"successful" : 5,
"failed" : 0
},
"hits" : {
"total" : 2,
"max_score" : 1.0,
"hits" : [ {
"_index" : "foo",
"_type" : "bar_full_name",
"_id" : "549c3ef5a0152464dde10bc4",
"_score" : 1.0,
"_source":{"_id":"549c3ef5a0152464dde10bc4","firstName":"Robert","fullName":"Robert Baratheon","lastName":"Baratheon"}
}, {
"_index" : "foo",
"_type" : "bar_full_name",
"_id" : "549c3f03a0152464dde10bc5",
"_score" : 1.0,
"_source":{"_id":"549c3f03a0152464dde10bc5","firstName":"John","fullName":"John Snow","lastName":"Snow"}
} ]
}
}
请注意字段 fullName，其中包含 firstName 和 lastName 之间的空格连接 - 我们的转换工作。
结论
现在我们知道如何使用 Transporter 将数据从 MongoDB 复制到 Elasticsearch，以及如何在同步时对我们的数据应用转换。 您可以以相同的方式应用更复杂的转换。 此外，您可以在管道中链接多个转换。
如果您正在执行多个转换，请将它们保存在单独的文件中并将它们链接起来，这是一个很好的做法。 这样，您就可以使您的每一个转换都可以在未来独立使用。
所以，差不多就是这样。 您可以查看 GitHub 上的 Transporter 项目，以随时了解 API 的最新更改。
您可能还想查看本教程，了解 Elasticsearch 中的基本 CRUD 操作。
