Константин Боенко

Ведущий разработчик ГК Юзтех

В последнее время всё большую популярность набирает Tarantool — платформа in-memory вычислений с гибкой схемой данных, включающая в себя NoSQL-базу данных и сервер приложений. В этой статье я хочу рассказать об одной из его реализаций — Tarantool Data Grid (TDG).

Что такое Tarantool Data Grid?

Tarantool Data Grid (TDG) — это, по сути, три компонента в одной коробке:

• Сам Tarantool

• Cartridge

• UI

Прежде чем работать c TDG, необходимо его настроить и сконфигурировать. Настройка осуществляется в UI и включает в себя настройку кластера Tarantool (собственно, это именно то, что делает Cartridge). Затем нужно загрузить в TDG набор файлов конфигурации, включающий в себя:

1. Модель (файл model.avsc) – обязательно;

2. Собственно файл конфигурации TDG (файл config.yml) – обязательно;

3. Пользовательские lua-скрипты – при необходимости.

Основой для работы с TDG является модель — местный эквивалент схемы данных реляционных БД. Модель TDG представляет собой avro-схему примерно следующего вида:

 [
  {
	"name": "User",
	"type": "record",
	"fields": [
  	{
    	"name": "user_id",
    	"type": "long"
  	},
  	{
    	"name": "name",
    	"type": "string"
  	},
  	{
    	"name": "age",
    	"type": [
                	"null",
                	"long"
            	]
  	},
  	{
    	"name": "sex",
    	"type": [
                	"null",
                	"boolean"
            	]
  	}
	],
	"indexes": [
  	"user_guid"
	]
  },
  {
	"name": "Address",
	"type": "record",
	"fields": [
  	{
 	   "name": "address_guid",
    	"type": "string"
  	},
  	{
    	"name": "city",
    	"type": "string"
  	},
  	  {
    	"name": "street",
    	"type": "string"
  	},
  	  {
    	"name": "house",
    	"type": "string"
  	},
  	{
    	"name": "apartment",
    	"type": [
                	"null",
                	"long"
            	]
  	}
	],
	"indexes": [
  	"address_guid"
	]
  }
]

Как видно, модель представляет собой avro-описание спейсов Tarantool, в которых будут храниться данные соответствующих типов. Спейсы будут созданы автоматически при загрузке модели в TDG. В нашем случае это спейсы User и Address.

У каждого спейса заданы индексы (в данном примере это guid). Но по желанию или необходимости индексы могут быть и составными. Для этого можно в модели написать конструкцию вида:

"indexes": [
  {
	"name": "primary",
	"parts": [
  	“field_1”,
  	"field_2",
  	…
	]
  }
]

Но модель — это далеко не всё. Обязательным элементом конфигурации TDG является корректно заполненный файл config.yml, который должен быть «отдан» TDG вместе с моделью. Заполнение config.yml — отдельная тема, всестороннее рассмотрение которой заслуживает отдельной статьи. Здесь мы ограничимся лишь отдельным примером, о котором я расскажу немного позже.

Для взаимодействия с TDG из приложений, разработанных на различных языках, существуют т.н. коннекторы. Есть коннектор и для Java + Spring, представляющий собой реализацию спецификации JPA — spring-data-tarantool. Для его подключения в файле pom.xml напишем его зависимость:

<dependency>
	<groupId>io.tarantool</groupId>
	<artifactId>spring-data-tarantool</artifactId>
	<version>0.5.2</version>
</dependency>

Пользоваться библиотекой довольно просто (особенно, если вы уже имели дело со Spring Data JPA).

Как обычно, описываем классы сущностей (сейчас ограничимся сущностью User из нашей модели):

@Tuple(“user”)
public class User  {
	@Id
	@Field("user_id")
	String id;
	String name;
	Integer age;
	@Field("sex")
	Boolean gender;
}

Как видите, аннотация @Field повешена не на все поля: в соответствии с принципами JPA маппинг этих полей будет производиться на основе полей класса сущности — а они совпадают с соответствующими полями модели. Там же, где имена полей не совпадают, ставим аннотацию.

Пишем репозиторий:

public interface UserRepository extends CrudRepository<User, Long> {
}

И всё, теперь мы можем обращаться к Tarantool так, как обычно и делаем, используя Spring Data JPA и стандартные методы jpa-репозитория. Например:

@Service
public class MyService {
	private final UserRepository repository;
	@Autowired
	public MyService(UserRepository repository) {
    	this.repository = repository;
	}
public void myMethod() {
    	User user = new User();
    	user.setName("Alexey Petrov");
    	user.setAge(28);
    	user.setGender(true)
    	User saved = repository.save(user);
    	List<User> userList = repository.findAll();
	}
}

Поставка TDG включает в себя коробочный модуль repository, в котором имеются реализации стандартных CRUD-методов. Методы JPA обращаются именно к ним. Мы также имеем возможность обратиться к методам модуля repository непосредственно. Для этого будем использовать аннотацию @Query. Например, для записи в спейс User мы можем использовать следующий метод:

@Query(function = "repository.put")
List<User> put(String typeName, User entity, Map<String, ?> options, List<?> context, Map<?, ?> credentials);

И тогда для записи очередного пользователя мы вызовем метод примерно так:

User myUser = …. // вызов метода создания юзера
List<User> saved = repository.put(“user”, myUser, new HashMap<>(), new HashMap<>())

Но что, если нам этого мало, и требуется какая-то дополнительная кастомная логика? И при этом желательно чтобы всё происходило в рамках одного обращения к TDG, а не нескольких? Здесь нам помогут функции, написанные на языке lua (именно этот язык выступает в качестве основного языка запросов Tarantool) и хранимые в TDG. Как уже говорилось выше, такие lua-скрипты загружаются в TDG вместе с конфигурационными файлами. 

Например, мы написали некую функцию (на всякий случай уточню: в lua это называется именно «функции», а не «методы») do_something, и поместили её в файл-модуль my_module.lua:

local log = require('log')
local function do_something(m)
  	log.info(‘Function do_something is called! M = ’ + m)
end
return {
  	do_something = do_something
}

Теперь мы хотим вызывать её в нашем Java-сервисе. Как это сделать?

Для этого нужно произвести дополнительное конфигурирование TDG.  Чтобы дать возможность вызывать нашу функцию извне, в файле config.yml напишем следующее:

services:
  do_something:
	doc: "test function"
	function: my_module.do_something
	return_type: any
	args:
  	m: string

Тем самым мы указываем, что снаружи можно вызвать из TDG сервис do_something. При его вызове будет вызвана функция do_something из файла my_module.lua, и эта функция принимает на вход строку, а возвращает что угодно (в нашем случае — ничего).

За вызов сервисов в TDG отвечает встроенный метод call_service. Воспользуемся им. В Java-сервисе в репозиторий добавим вот такой метод:

@Query(function = "call_service")
List<T> callService(String name, Map<String, ?> args, Map<?, ?> options);

И теперь в нужном месте мы можем вызвать наш сервис следующим образом:

public void callTDGService(String str) {
   Map<String, String> args = new HashMap<>();
   args.put(“m”, “hello, TDG!”)
	service.callService("do_something", args, new HashMap<>());
}

Но и это еще не всё. TDG позволяет осуществить последовательный «конвейерный» вызов, и, таким образом, одномоментно выполнять такие действия, как приём, обработка, сохранение и репликация входных данных. Для этого в TDG существует функция tarantool_protocol_process. Вызывается она точно так же, как и любая другая функция TDG.  Пишем в репозитории метод:

@Query(function = "tarantool_protocol_process")
String process(RawUser user, Map<?, ?> options);

Предположим, что мы хотим отправлять в TDG инстанс некоего класса RawUser, там в соответствии с какой-то логикой преобразовывать его в уже знакомый нам класс User, сохранять его в соответствующий спейс, после чего реплицировать эти данные в имеющееся у нас резервное хранилище (например, БД PostgreSQL). Здесь-то нам и поможет функция tarantool_protocol_process. Она принимает наш объект и помещает его в конвейерную очередь, где он будет обрабатываться согласно тому, как мы эту очередь сконфигурируем. А значит, приступим к конфигурации. Снова открываем наш файл config.yml и напишем там следующее:

connector:
  input:
	- name: tarantool_protocol
  	type: tarantool_protocol
  	routing_key: user
  output:
	- name: user
  	type: dummy
input_processor:
  handlers:
	- key: user
  	function: my_module.create_user_from_raw_user
  storage:
	- key: user
  	type: User
output_processor:
  user:
	handlers:
  	- function: pg_replicator.save_user
    	outputs:
      	- user
odbc:
  - dsn: DRIVER=/usr/pgsql-13/lib/psqlodbc.so;SERVER=<url>;PORT=<port>;DATABASE=<db-name>;UID=<user>;PWD=<password>;
	name: postgres

Сначала мы настроили коннекторы для получения (input) и для возвращения (output) данных.  Здесь есть очень важный момент – routing_key. Именно по нему очередь будет отличать ваши данные от всех остальных. Этот ключ можно задать самим первым действием, когда объект попадет в очередь. Но мы ограничимся тем, который будет ему назначен по умолчанию (в нашем случае это user).

Далее конфигурируем input_processor, т.е. действия, которые будут производиться с объектом на этапе получения. В секции handlers описано, что при получении данных с ключом user будет вызвана функция create_user_from_raw_user в модуле my_module.lua. Представим, что мы написали такую функцию, и она возвращает объект типа User (в соответствии с загруженной в TGD моделью).

Далее попадаем в секцию storage. В ней говорится, что объект с ключом user будет сохранен в спейc User. При этом, если такой записи в спейсе нет, она будет добавлена. Если она уже есть, то она будет обновлена.

Далее следует конфигурирование output_processor – действий, которые будут совершены с объектом перед удалением из очереди. В нашем случает это репликация во внешнюю систему – БД PostgreSQL.  Мы описываем, что при попадании в output_processor данный с ключом user будет вызвана функция save_user в модуле pg_replicator.lua. А чтобы она отработала корректно, мы должны сконфигурировать подключение к PostrgeSQL в секции odbc.

Листинг файла pg_replicator.lua
local odbc = require(‘odbc’)
local function save_user(user)
  	local res, err = odbc.execute(‘postgres’, “insert into my_scheme.user(name, age, sex) values (?, ?, ?)”,
              	{user.name, user.age, user.sex})
return res
end

Вот так всё просто.