Синхронизация преобразованных данных между MongoDB и Elasticsearch с помощью Transporter в Ubuntu 14.04

Elasticsearch упрощает полнотекстовый поиск по данным, а MongoDB – хранение данных. эти приложения часто используются в связке.

Transporter – это инструмент Compose с открытым исходным кодом для перемещения данных между разными хранилищами данных. Разработчики часто пишут одноразовые скрипты для выполнения таких задач, как перемещение данных между несколькими базами данных, перемещение данных из файлов в БД или наоборот. Но использование такого инструмента, как Transporter, имеет ряд преимуществ.

Transporter позволяет создавать конвейеры, которые определяют поток из источника (source, откуда читаются данные) в приемник данных (sink, куда записываются данные). Источниками и приемниками могут быть базы данных SQL или NoSQL, плоские файлы и другие ресурсы.

Кроме перемещения данных Transporter также позволяет изменять данные при прохождении через конвейер с помощью трансформаторов.

В этом мануале мы рассмотрим пример перемещения и обработки данных из БД MongoDB в Elasticsearch с помощью Transporter в Ubuntu 14.04.

Читайте также: Синхронизация преобразованных данных между MongoDB и Elasticsearch с помощью Transporter в Ubuntu 16.04

Требования

Для работы вам нужен сервер Ubuntu 14.04, настроенный по этому мануалу.

1: Установка MongoDB

Импортируйте открытый ключ репозитория MongoDB:

sudo apt-key adv --keyserver hkp://keyserver.ubuntu.com:80 --recv 7F0CEB10

Создайте list-файл для 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

Создайте новый каталог для установки программы и перейдите в него:

mkdir ~/utils
cd ~/utils

На сайте Elasticsearch найдите нужную вам версию (лучше использовать последнюю).

Загрузите выбранную версию:

wget https://download.elasticsearch.org/elasticsearch/elasticsearch/elasticsearch-1.5.0.zip

Установите unzip:

sudo apt-get install unzip

Распакуйте архив:

unzip elasticsearch-1.5.0.zip

Перейдите в полученный каталог:

cd elasticsearch-1.5.0

Запустите Elasticsearch:

bin/elasticsearch

На это уйдет пара секунд. После этого Elasticsearch запустится в терминале.

Примечание: Возможно, вы захотите запустить Elasticsearch как сервис, чтобы иметь доступ к команде sudo service elasticsearch restart. Больше информации вы найдете в мануале Система инициализации Upstart: что это такое и как это работает. Также вы можете установить Elasticsearch из репозитория Ubuntu.

Оставьте этот терминал открытым. Создайте новое SSH-подключение в новом окне терминала и убедитесь, что экземпляр работает:

curl -XGET http://localhost:9200

По умолчанию Elasticsearch использует порт 9200. Если все работает правильно, вы увидите такой вывод:

{
"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"
}

4: Установка Mercurial

Теперь нужно установить инструмент управления версиями Mercurial.

sudo apt-get install mercurial

Убедитесь, что установка прошла успешно:

hg
Mercurial Distributed SCM
basic commands:
. . .

5: Установка Go

Transporter написан на языке Go. Потому вам нужно установить golang.

sudo apt-get install golang

Чтобы Go работал правильно, нужно установить следующие переменные среды.

Создайте каталог для Go в $HOME:

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/8host/go

Если путь отображается неправильно, просмотрите предыдущие разделы мануала, чтобы найти ошибку.

Установив $GOPATH, нужно убедиться, что Go установлен правильно. Для этого соберите простую программу.

Создайте файл hello.go и поместите в него следующую программу.

nano ~/hello.go

Скопируйте и вставьте короткую программу «Hello, world».

package main;
import "fmt"
func main() {
fmt.Printf("Hello, world\n")
}

Нажмите CTRL+X, чтобы закрыть, и Y+ENTER, чтобы сохранить файл.

В домашнем каталоге запустите файл Go:

go run hello.go
Hello, world

Если вы видите такое сообщение, установка Go прошла успешно.

Перейдите в каталог $GOPATH и создайте подкаталоги src, pkg и bin. Они образуют рабочее пространство Go.

cd $GOPATH
mkdir src pkg bin

• src хранит исходные файлы Go, собранные в пакеты (по пакету на каждый каталог).
• pkg содержит объекты пакетов.
• bin содержит исполняемые команды.

6: Установка Git

Git понадобится для установки Transporter:

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.

Клонируйте GitHub репозиторий.

git clone https://github.com/compose/transporter.git

Перейдите в новый каталог:

cd transporter

Передайте права на каталог /usr/lib/go:

sudo chown -R $USER /usr/lib/go

Убедитесь, что build-essential установлен в систему:

sudo apt-get install build-essential

Чтобы загрузить все зависимости, запустите:

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:

• База данных – это аналог индексов в Elasticsearch.
• Коллекция – аналог типов в Elasticsearch.

Теперь нужно синхронизировать данные коллекции bar БД foo из MongoDB в тип bar индекса foo в Elasticsearch.

9: Настройка Transporter

Теперь нужно настроить синхронизацию данных. Transporter нужен конфигурационный файл config.yaml, трансформатор myTransformation.js и файл приложения application.js.

• В конфигурационном файле хранятся данные о нодах, типах и URI.
• В файле приложения определяется поток данных, его источник и приемник, а также опциональные параметры трансформации.
• Трансформатор определяет, как будут преобразовываться данные.

Примечание: Все команды в данном разделе нужно выполнить в каталоге transporter.

Перейдите в каталог transporter:

cd ~/go/src/github.com/compose/transporter

Конфигурационный файл

Создайте резервную копию файла config.yaml и добавьте в новый файл свои параметры.

mv test/config.yaml test/config.yaml.00

Этот файл не будет сильно отличаться, но в нем будут дополнительные URI и другие параметры. Скопируйте содержимое файла и вставьте его в config.yaml.

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. Ноды – это различные источники и приемники данных. type – это тип ноды.

• mongo указывает на экземпляр MongoDB.
• elasticsearch указывает на ноду Elasticsearch.
• file значит, что это простой текстовый файл.

uri предоставит конечную точку API для соединения с нодой. MongoDB будет использовать порт по умолчанию (27017), если не указан другой порт. Поскольку нам нужно извлечь данные из базы данных foo MongoDB, URI должен выглядеть так:

mongodb://localhost/foo

URI Elasticsearch имеет такой вид:

http://localhost:9200/

Сохраните файл config.yaml.

Файл приложения

Откройте 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(). Вторая строка объединяет 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

Команда 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 на GitHub