[Перевод] Apache Ignite: распределенные вычисления в оперативной памяти
Привет, Хабр!
Мы продолжаем интересоваться новыми решениями от компании Apache. Рассчитываем выпустить в мае книгу «High Performance Spark» Холдена Карау (книга в верстке), а в августе — книгу «Kafka: The Definitive Guide» Нии Нархид (еще в переводе). Сегодня же хотим предложить краткую ознакомительную статью об Apache Ignite и оценить масштаб интереса к теме.
Приятного чтения!
Apache Ignite — относительно новое решение, однако, популярность его быстро растет. Сложно отнести его к какому-то конкретному подвиду движков баз данных, поскольку характеристики Ignite роднят его с несколькими инструментами. Основное назначение этого инструмента — хранение распределенных данных в оперативной памяти, а также хранение информации в формате «ключ-значение». Также в Ignite есть некоторые общие функции RDBMS, в частности, поддержка SQL-запросов и ACID-транзакций. Но это не означает, что данное решение — типичная база данных для работы с транзакциями на языке SQL. Здесь не поддерживаются ограничения внешнего ключа, а транзакции доступны лишь на уровне «ключ-значение». Тем не менее, Apache Ignite кажется очень интересным решением.
Apache Ignite легко запустить как узел, встроенный в приложение Spring Boot. Проще всего этого добиться при помощи библиотеки Spring Data Ignite. Apache Ignite реализует интерфейс Spring Data CrudRepository, поддерживающий основные операции CRUD, а также обеспечивающий доступ к гриду Apache Ignite SQL Grid с использованием унифицированных интерфейсов Spring Data. Хотя, в нем обеспечивается сохраняемость данных в дисковом хранилище с поддержкой SQL и парадигмы ACID, мы разработали решение для сохранения объектов кэша оперативной памяти в базу данных MySQL. Архитектура предлагаемого решения показана на рисунке ниже — как видите, она очень проста. Приложение помещает данные в кэш оперативной памяти, устроенный в Apache Ignite. Apache Ignite автоматически синхронизирует эти изменения с базой данных в ходе асинхронной фоновой задачи. Способ считывания данных в этом приложении также не должен вас удивить. Если сущность не кэширована, то она считывается из базы данных и помещается в кэш на будущее.
Здесь я подробно опишу, как разрабатывается приложение такого рода. Результат выложен на GitHub. Я нашел в Интернете еще несколько примеров, но в них затронуты только основы. Я покажу, как сконфигурировать Apache Ignite для записи объектов из кэша в базу данных, а также как создавать более сложные запросы на объединение с использованием нескольких кэшей. Начнем с запуска базы данных.
1. Настраиваем базу данных MySQL
Чтобы запустить базу данных MySQL локально, лучше всего, конечно же, воспользоваться контейнером Docker. База данных MySQL для Docker под Windows в настоящее время доступна по адресу 192.168.99.100:33306.
docker run -d --name mysql -e MYSQL_DATABASE=ignite -e MYSQL_USER=ignite -e MYSQL_PASSWORD=ignite123 -e MYSQL_ALLOW_EMPTY_PASSWORD=yes -p 33306:3306 mysql
Далее создаем таблицы, используемые сущностями приложения для хранения данных: PERSON, CONTACT. Они относятся к таблицам как 1…N, где в таблице CONTACT содержится внешний ключ, указывающий на PERSON id.
CREATE TABLE `person` (
`id` int(11) NOT NULL,
`first_name` varchar(45) DEFAULT NULL,
`last_name` varchar(45) DEFAULT NULL,
`gender` varchar(10) DEFAULT NULL,
`country` varchar(10) DEFAULT NULL,
`city` varchar(20) DEFAULT NULL,
`address` varchar(45) DEFAULT NULL,
`birth_date` date DEFAULT NULL,
PRIMARY KEY (`id`)
);
CREATE TABLE `contact` (
`id` int(11) NOT NULL,
`location` varchar(45) DEFAULT NULL,
`contact_type` varchar(10) DEFAULT NULL,
`person_id` int(11) NOT NULL,
PRIMARY KEY (`id`)
);
ALTER TABLE `ignite`.`contact` ADD INDEX `person_fk_idx` (`person_id` ASC);
ALTER TABLE `ignite`.`contact`
ADD CONSTRAINT `person_fk` FOREIGN KEY (`person_id`) REFERENCES `ignite`.`person` (`id`) ON DELETE CASCADE ON UPDATE CASCADE;2. Конфигурируем Maven
Чтобы приступить к работе с репозиторием Spring Data для Apache Ignite, проще всего добавить следующую зависимость Maven в файл pom.xml нашего приложения. Все остальные зависимости Ignite будут включены автоматически. Нам также понадобится драйвер MySQL JDBC driver и зависимости Spring JDBC, чтобы сконфигурировать соединение с базой данных. Они необходимы, так как мы встраиваем Apache Ignite в приложение, и требуется подключиться к базе данных MySQL, чтобы можно было синхронизировать кэш с таблицами базы данных.
org.springframework.boot
spring-boot-starter-web
org.springframework.boot
spring-boot-starter-jdbc
mysql
mysql-connector-java
runtime
org.apache.ignite
ignite-spring-data
${ignite.version}
3. Конфигурируем узел Ignite
Класс IgniteConfiguration позволяет сконфигурировать все доступные настройки узла Ignite. В данном случае наиболее важна конфигурация кэша (1). Следует добавить главный ключ и классы сущностей как индексированные типы (2). Далее нужно предусмотреть экспорт обновлений кэша в базу данных (3) и считывать из базы данных ту информацию, которой не окажется в кэше (4). Взаимодействие между узлом Ignite и MySQL можно сконфигурировать при помощи класса CacheJdbcPojoStoreFactory (5). Там нужно передать DataSource @Bean (6), диалект (7) и соответствие между полями объекта и столбцами таблицы (8).
@Bean
public Ignite igniteInstance() {
IgniteConfiguration cfg = new IgniteConfiguration();
cfg.setIgniteInstanceName("ignite-1");
cfg.setPeerClassLoadingEnabled(true);
CacheConfiguration ccfg2 = new CacheConfiguration<>("ContactCache"); // (1)
ccfg2.setIndexedTypes(Long.class, Contact.class); // (2)
ccfg2.setWriteBehindEnabled(true);
ccfg2.setWriteThrough(true); // (3)
ccfg2.setReadThrough(true); // (4)
CacheJdbcPojoStoreFactory f2 = new CacheJdbcPojoStoreFactory<>(); // (5)
f2.setDataSource(datasource); // (6)
f2.setDialect(new MySQLDialect()); // (7)
JdbcType jdbcContactType = new JdbcType(); // (8)
jdbcContactType.setCacheName("ContactCache");
jdbcContactType.setKeyType(Long.class);
jdbcContactType.setValueType(Contact.class);
jdbcContactType.setDatabaseTable("contact");
jdbcContactType.setDatabaseSchema("ignite");
jdbcContactType.setKeyFields(new JdbcTypeField(Types.INTEGER, "id", Long.class, "id"));
jdbcContactType.setValueFields(new JdbcTypeField(Types.VARCHAR, "contact_type", ContactType.class, "type"), new JdbcTypeField(Types.VARCHAR, "location", String.class, "location"), new JdbcTypeField(Types.INTEGER, "person_id", Long.class, "personId"));
f2.setTypes(jdbcContactType);
ccfg2.setCacheStoreFactory(f2);
CacheConfiguration ccfg = new CacheConfiguration<>("PersonCache");
ccfg.setIndexedTypes(Long.class, Person.class);
ccfg.setWriteBehindEnabled(true);
ccfg.setReadThrough(true);
ccfg.setWriteThrough(true);
CacheJdbcPojoStoreFactory f = new CacheJdbcPojoStoreFactory<>();
f.setDataSource(datasource);
f.setDialect(new MySQLDialect());
JdbcType jdbcType = new JdbcType();
jdbcType.setCacheName("PersonCache");
jdbcType.setKeyType(Long.class);
jdbcType.setValueType(Person.class);
jdbcType.setDatabaseTable("person");
jdbcType.setDatabaseSchema("ignite");
jdbcType.setKeyFields(new JdbcTypeField(Types.INTEGER, "id", Long.class, "id"));
jdbcType.setValueFields(new JdbcTypeField(Types.VARCHAR, "first_name", String.class, "firstName"), new JdbcTypeField(Types.VARCHAR, "last_name", String.class, "lastName"), new JdbcTypeField(Types.VARCHAR, "gender", Gender.class, "gender"), new JdbcTypeField(Types.VARCHAR, "country", String.class, "country"), new JdbcTypeField(Types.VARCHAR, "city", String.class, "city"), new JdbcTypeField(Types.VARCHAR, "address", String.class, "address"), new JdbcTypeField(Types.DATE, "birth_date", Date.class, "birthDate"));
f.setTypes(jdbcType);
ccfg.setCacheStoreFactory(f);
cfg.setCacheConfiguration(ccfg, ccfg2);
return Ignition.start(cfg);
}
Вот конфигурация источника данных Spring для MySQL в виде контейнера Docker.
spring:
datasource:
name: mysqlds
url: jdbc:mysql://192.168.99.100:33306/ignite?useSSL=false
username: ignite
password: ignite123
Здесь необходимо отметить, что Apache Ignite не лишен некоторых недостатков. Например, он отображает Enum на целое число и берет его порядковое значение, хотя, конфигурирует VARCHAR как тип JDCB. Когда такой ряд считывается из базы данных, он неверно отображается на Enum в объекте — в этом поле отклика у вас получится null.
new JdbcTypeField(Types.VARCHAR, "contact_type", ContactType.class, "type")4. Объекты модели
Как упоминалось выше, в схеме нашей базы данных две таблицы. Еще есть два класса модели и две конфигурации кэша, по одной на каждый класс модели. Ниже приведена реализация класса модели. Одна из самых интересных вещей, которую здесь стоит отметить — генерация ID при помощи класса AtomicLong. Это один из базовых компонентов Ignite, который служит генератором последовательностей. Также видим специфичную аннотацию @QuerySqlField; если она сопровождает поле — это означает, что данное поле может использоваться в SQL как параметр запроса.
@QueryGroupIndex.List(
@QueryGroupIndex(name="idx1")
)
public class Person implements Serializable {
private static final long serialVersionUID = -1271194616130404625L;
private static final AtomicLong ID_GEN = new AtomicLong();
@QuerySqlField(index = true)
private Long id;
@QuerySqlField(index = true)
@QuerySqlField.Group(name = "idx1", order = 0)
private String firstName;
@QuerySqlField(index = true)
@QuerySqlField.Group(name = "idx1", order = 1)
private String lastName;
private Gender gender;
private Date birthDate;
private String country;
private String city;
private String address;
private List contacts = new ArrayList<>();
public void init() {
this.id = ID_GEN.incrementAndGet();
}
public Long getId() {
return id;
}
public void setId(Long id) {
this.id = id;
}
public String getFirstName() {
return firstName;
}
public void setFirstName(String firstName) {
this.firstName = firstName;
}
public String getLastName() {
return lastName;
}
public void setLastName(String lastName) {
this.lastName = lastName;
}
public Gender getGender() {
return gender;
}
public void setGender(Gender gender) {
this.gender = gender;
}
public Date getBirthDate() {
return birthDate;
}
public void setBirthDate(Date birthDate) {
this.birthDate = birthDate;
}
public String getCountry() {
return country;
}
public void setCountry(String country) {
this.country = country;
}
public String getCity() {
return city;
}
public void setCity(String city) {
this.city = city;
}
public String getAddress() {
return address;
}
public void setAddress(String address) {
this.address = address;
}
public List getContacts() {
return contacts;
}
public void setContacts(List contacts) {
this.contacts = contacts;
}
} 5. Репозитории Ignite
Полагаю, вам известно, как в Spring Data JPA создаются репозитории. Обработка репозиториев должна быть обеспечена в классе main или @Configuration.
@SpringBootApplication
@EnableIgniteRepositories
public class IgniteRestApplication {
@Autowired
DataSource datasource;
public static void main(String[] args) {
SpringApplication.run(IgniteRestApplication.class, args);
}
// ...
}
Затем расширяем наш интерфейс @Repository базовым интерфейсом CrudRepository. Он поддерживает только унаследованные методы с параметром id. В приведенном ниже фрагменте PersonRepository я определил несколько поисковых методов, воспользовавшись соглашениями об именовании, принятыми в v Spring Data, и запросами Ignite. Эти примеры демонстрируют, что можно вернуть в результатах запроса либо полный объект, либо избранные поля из него — в зависимости от того, что нам требуется.
@RepositoryConfig(cacheName = "PersonCache")
public interface PersonRepository extends IgniteRepository {
List findByFirstNameAndLastName(String firstName, String lastName);
@Query("SELECT c.* FROM Person p JOIN \"ContactCache\".Contact c ON p.id=c.personId WHERE p.firstName=? and p.lastName=?")
List selectContacts(String firstName, String lastName);
@Query("SELECT p.id, p.firstName, p.lastName, c.id, c.type, c.location FROM Person p JOIN \"ContactCache\".Contact c ON p.id=c.personId WHERE p.firstName=? and p.lastName=?")
List> selectContacts2(String firstName, String lastName);
}
6. API и тестирование
Вот теперь можно внедрить компоненты репозитория в классы контроллеров REST. API предоставит методы для добавления новых объектов в кэш, обновления или удаления имеющихся объектов, а также для поиска по первичному ключу или по другим, более сложным индексам.
@RestController
@RequestMapping("/person")
public class PersonController {
private static final Logger LOGGER = LoggerFactory.getLogger(PersonController.class);
@Autowired
PersonRepository repository;
@PostMapping
public Person add(@RequestBody Person person) {
person.init();
return repository.save(person.getId(), person);
}
@PutMapping
public Person update(@RequestBody Person person) {
return repository.save(person.getId(), person);
}
@DeleteMapping("/{id}")
public void delete(Long id) {
repository.delete(id);
}
@GetMapping("/{id}")
public Person findById(@PathVariable("id") Long id) {
return repository.findOne(id);
}
@GetMapping("/{firstName}/{lastName}")
public List findByName(@PathVariable("firstName") String firstName, @PathVariable("lastName") String lastName) {
return repository.findByFirstNameAndLastName(firstName, lastName);
}
@GetMapping("/contacts/{firstName}/{lastName}")
public List findByNameWithContacts(@PathVariable("firstName") String firstName, @PathVariable("lastName") String lastName) {
List persons = repository.findByFirstNameAndLastName(firstName, lastName);
List contacts = repository.selectContacts(firstName, lastName);
persons.stream().forEach(it -> it.setContacts(contacts.stream().filter(c -> c.getPersonId().equals(it.getId())).collect(Collectors.toList())));
LOGGER.info("PersonController.findByIdWithContacts: {}", contacts);
return persons;
}
@GetMapping("/contacts2/{firstName}/{lastName}")
public List findByNameWithContacts2(@PathVariable("firstName") String firstName, @PathVariable("lastName") String lastName) {
List> result = repository.selectContacts2(firstName, lastName);
List persons = new ArrayList<>();
for (List l : result) {
persons.add(mapPerson(l));
}
LOGGER.info("PersonController.findByIdWithContacts: {}", result);
return persons;
}
private Person mapPerson(List l) {
Person p = new Person();
Contact c = new Contact();
p.setId((Long) l.get(0));
p.setFirstName((String) l.get(1));
p.setLastName((String) l.get(2));
c.setId((Long) l.get(3));
c.setType((ContactType) l.get(4));
c.setLocation((String) l.get(4));
p.addContact(c);
return p;
}
}
Конечно же, важно проверить производительность созданного решения, особенно когда оно связано с хранением распределенных данных в оперативной памяти и с базами данных. Для этого я написал несколько junit-тестов, помещающих в кэш большое количество объектов, а затем вызывающих методы поиска (для ввода используются случайные данные) — так проверяется производительность запросов. Вот метод, генерирующий много объектов Person и Contact и помещающий их в кэш при помощи API конечных точек.
@Test
public void testAddPerson() throws InterruptedException {
ExecutorService es = Executors.newCachedThreadPool();
for (int j = 0; j < 10; j++) { es.execute(() -> {
TestRestTemplate restTemplateLocal = new TestRestTemplate();
Random r = new Random();
for (int i = 0; i < 1000000; i++) {
Person p = restTemplateLocal.postForObject("http://localhost:8090/person", createTestPerson(), Person.class);
int x = r.nextInt(6);
for (int k = 0; k < x; k++) {
restTemplateLocal.postForObject("http://localhost:8090/contact", createTestContact(p.getId()), Contact.class);
}
}
});
}
es.shutdown();
es.awaitTermination(60, TimeUnit.MINUTES);
}
В Spring Boot предоставляются методы для взятия основных характеристик, позволяющих судить о скорости отклика API. Чтобы активировать такую возможность, нужно включить в зависимости Spring Actuator. Конечная точка Metrics доступна по адресу localhost:8090/metrics. Она не только показывает, сколько времени затрачивает на работу каждый метод API, но и выводит статистику по таким показателям, как количество действующих потоков или свободная память.
7. Запуск приложения
Теперь запустим получившееся у нас приложение, в которое встроен узел Apache Ignite. Я учел советы по повышению производительности, которые содержатся в документации Ignite и определил конфигурацию JVM, показанную ниже.
java -jar -Xms512m -Xmx1024m -XX:MaxDirectMemorySize=256m -XX:+DisableExplicitGC -XX:+UseG1GC target/ignite-rest-service-1.0-SNAPSHOT.jar
Теперь можно запустить тестовый класс JUnit IgniteRestControllerTest. Он помещает в кэш некоторое количество данных, а затем вызывает методы поиска. Даны параметры для тестов, где в кэше использованы 1M объектов Person и 2.5M объектов Contact. Каждый из методов поиска выполняется в среднем за 1 мс.
{
"mem": 624886,
"mem.free": 389701,
"processors": 4,
"instance.uptime": 2446038,
"uptime": 2466661,
"systemload.average": -1,
"heap.committed": 524288,
"heap.init": 524288,
"heap.used": 133756,
"heap": 1048576,
"threads.peak": 107,
"threads.daemon": 25,
"threads.totalStarted": 565,
"threads": 80,
...
"gauge.response.person.contacts.firstName.lastName": 1,
"gauge.response.contact": 1,
"gauge.response.person.firstName.lastName": 1,
"gauge.response.contact.location.location": 1,
"gauge.response.person.id": 1,
"gauge.response.person": 0,
"counter.status.200.person.id": 1000,
"counter.status.200.person.contacts.firstName.lastName": 1000,
"counter.status.200.person.firstName.lastName": 1000,
"counter.status.200.contact": 2500806,
"counter.status.200.person": 1000000,
"counter.status.200.contact.location.location": 1000
}