ПРИМЕЧАНИЕ: Интеграция с Fluent Bit требует наличия Manticore Buddy. Если она не работает, убедитесь, что Buddy установлен.

Vector от Datadog — это открытый конвейер данных для наблюдаемости, который может собирать, преобразовывать и маршрутизировать логи или метрики. Хотя Vector может агрегировать данные самостоятельно, его сочетание с Manticore предоставляет специализированный слой для хранения и поиска.

Пример ниже показывает, как перенаправлять dpkg.log Debian через Vector.dev и индексировать его в Manticore.

2023-05-31 10:42:55 status triggers-awaited ca-certificates-java:all 20190405ubuntu1.1
2023-05-31 10:42:55 trigproc libc-bin:amd64 2.31-0ubuntu9.9 <none>
2023-05-31 10:42:55 status half-configured libc-bin:amd64 2.31-0ubuntu9.9
2023-05-31 10:42:55 status installed libc-bin:amd64 2.31-0ubuntu9.9
2023-05-31 10:42:55 trigproc systemd:amd64 245.4-4ubuntu3.21 <none>

Создайте vector.toml, аналогичный следующему:

[sources.test_file]
type = "file"
include = [ "/var/log/dpkg.log" ]

[transforms.modify_test_file]
type = "remap"
inputs = [ "test_file" ]
source = """
.vec_timestamp = del(.timestamp)"""

[sinks.manticore]
type = "elasticsearch"
inputs = [ "modify_test_file" ]
endpoints = ["http://127.0.0.1:9308"]
bulk.index = "dpkg_log"

• endpoints указывает на HTTP-интерфейс Manticore (порт 9308 по умолчанию). Измените его, если ваш экземпляр слушает на другом порту.
• Преобразование remap перемещает поле timestamp по умолчанию Vector в vec_timestamp, потому что timestamp зарезервировано в Manticore.
• bulk.index определяет таблицу, которая будет создана автоматически, когда Vector начнёт отправлять данные.

Запустите Vector.dev с этой конфигурацией, и он будет отслеживать файл лога, преобразовывать каждое событие и перенаправлять его напрямую в Manticore.

Сохраните конфигурацию как vector.toml, затем запустите агент:

vector --config vector.toml

Если вы запускаете Vector.dev в Docker, подключите как файл конфигурации, так и каталог с логами, например:

docker run --rm -v /var/log/dpkg.log:/var/log/dpkg.log:ro \
  -v $(pwd)/vector.toml:/etc/vector/vector.toml:ro \
  timberio/vector:latest --config /etc/vector/vector.toml

Когда конвейер работает, Manticore автоматически создаёт таблицу dpkg_log. Её схема и пример документов выглядят так:

mysql> DESCRIBE dpkg_log;
+-----------------+---------+--------------------+
| Field           | Type    | Properties         |
+-----------------+---------+--------------------+
| id              | bigint  |                    |
| file            | text    | indexed stored     |
| host            | text    | indexed stored     |
| message         | text    | indexed stored     |
| source_type     | text    | indexed stored     |
| vec_timestamp   | text    | indexed stored     |
+-----------------+---------+--------------------+

mysql> SELECT * FROM dpkg_log LIMIT 3\G
*************************** 1. row ***************************
id: 7856533729353672195
file: /var/log/dpkg.log
host: logstash-787f68f6f-nhdd2
message: 2023-06-05 14:03:04 startup archives install
source_type: file
vec_timestamp: 2023-08-04T15:32:50.203091741Z
*************************** 2. row ***************************
id: 7856533729353672196
file: /var/log/dpkg.log
host: logstash-787f68f6f-nhdd2
message: 2023-06-05 14:03:04 install base-passwd:amd64 <none> 3.5.47
source_type: file
vec_timestamp: 2023-08-04T15:32:50.203808861Z
*************************** 3. row ***************************
id: 7856533729353672197
file: /var/log/dpkg.log
host: logstash-787f68f6f-nhdd2
message: 2023-06-05 14:03:04 status half-installed base-passwd:amd64 3.5.47
source_type: file
vec_timestamp: 2023-08-04T15:32:50.203814031Z

Использование Vector.dev с Manticore позволяет собирать логи практически из любого источника, обогащать или очищать их на лету и сохранять результаты в базе данных, готовой к поиску. Этот рабочий процесс сохраняет конвейеры наблюдаемости простыми, но при этом позволяет выполнять расширенные преобразования, когда это необходимо.

ПРИМЕЧАНИЕ: Некоторые панели и метрики на дашбордах требуют Manticore Buddy. Если некоторые панели остаются пустыми, убедитесь, что Buddy установлен.

Grafana — это платформа с открытым исходным кодом для визуализации данных и мониторинга, которая позволяет создавать интерактивные дашборды и диаграммы. Manticore Search может интегрироваться с Grafana двумя основными способами:

1. Мониторинг производительности Manticore: Использование Prometheus для сбора метрик от Manticore и их визуализации в дашбордах Grafana. Этот подход фокусируется на состоянии системы, мониторинге производительности и оповещениях для самого поискового движка.
2. Визуализация поисковых данных: Подключение через коннектор MySQL для запросов и отображения данных, хранящихся в таблицах Manticore, аналогично тому, как Kibana используется для визуализации данных Elasticsearch. Это идеально подходит для анализа тенденций, агрегаций и создания пользовательских визуализаций на основе ваших индексированных данных.

В настоящее время протестированы и поддерживаются версии Grafana 10.0-12.4.

Manticore предоставляет предварительно собранный дашборд Grafana и правила оповещений Prometheus в специальном репозитории manticoresoftware/grafana-dashboard. Репозиторий также включает в себя универсальный Docker-образ с Grafana и Prometheus для быстрого старта. Эти ресурсы можно использовать независимо с вашей существующей настройкой Grafana и Prometheus, что позволяет отслеживать операционные метрики Manticore — такие как задержка, использование ресурсов и ошибки — без необходимости запускать отдельный экземпляр.

• Дополнительные скриншоты: https://github.com/manticoresoftware/grafana-dashboard
• JSON дашборда: grafana/dashboards/manticore-dashboard.json
• Правила оповещений (Prometheus/Alertmanager): prometheus/rules/manticore-alerts.yml

Для полностью предварительно настроенной конфигурации запустите единый Docker-образ с Grafana и Prometheus, адаптированный для мониторинга Manticore. Этот вариант идеален, если у вас нет существующей настройки или вы хотите быстро создать тестовое окружение:

docker run -e MANTICORE_TARGETS=localhost:9308 -p 127.0.0.1:3000:3000 manticoresearch/dashboard:latest

MANTICORE_TARGETS — это список HTTP-эндпоинтов Manticore, разделенных запятыми, в формате хост:порт (например: localhost:9308,10.0.0.5:9308).

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

Чтобы добавить дашборд Manticore в вашу существующую инсталляцию Grafana:

1. В Grafana перейдите в Дашборды → Создать → Импортировать.
2. Загрузите файл manticore-dashboard.json (или вставьте его JSON-содержимое).
3. При запросе выберите ваш источник данных Prometheus.
4. При необходимости проверьте и настройте переменную дашборда job (по умолчанию ожидается значение метки manticoresearch).

Для интеграции правил оповещений Manticore в вашу существующую систему Prometheus:

1. Скачайте файл manticore-alerts.yml и добавьте его в конфигурацию rule_files вашего Prometheus.
2. Если имя вашего задания для сбора метрик (scrape job) отличается от manticoresearch, обновите селекторы {job="manticoresearch"} в правилах соответствующим образом.
3. Перезагрузите или перезапустите Prometheus, чтобы применить изменения.

Файл manticore-alerts.yml разработан для отслеживания ключевых сигналов, связанных с доступностью, перегрузкой и истощением ресурсов — проблем, которые часто напрямую влияют на пользователей. Пороговые значения установлены как безопасные по умолчанию; настройте их в соответствии с вашей конкретной нагрузкой.

• ManticoreTargetDown (critical): Prometheus не может собрать метрики Manticore (up == 0 в течение 2 минут). Обычно это означает, что конечная точка метрик недоступна, недостижима или конфигурация сбора неверна.
• ManticoreRecentlyRestarted (warning): Время работы остается ниже 5 минут в течение 5 минут. Часто сигнализирует о нестабильности, такой как циклы сбоев, убийства из-за нехватки памяти (OOM) или перезапуски оркестратором.
• ManticoreMaxedOutErrors (warning): manticore_maxed_out_error_count увеличивается за последние 5 минут. Manticore отклоняет запросы из-за достижения пределов емкости, параллелизма или ресурсов.
• ManticoreSearchLatencyP95High (warning): P95 задержка поиска превышает 500 мс в течение 10 минут. Многие пользователи могут воспринимать поиск как медленный.
• ManticoreSearchLatencyP99High (critical): P99 задержка поиска превышает 1000 мс в течение 10 минут. Хвостовая задержка сильно повышена, что указывает на очень медленные запросы в наихудшем случае.
• ManticoreWorkQueueBacklog (warning): Длина очереди задач превышает 100 в течение 5 минут. Запросы накапливаются, что часто приводит к увеличению задержки.
• ManticoreWorkersSaturated (warning): Более 90% рабочих потоков активны в течение 10 минут. Рабочие потоки приближаются к полной загрузке, что, вероятно, вызывает более высокую задержку и очередь.
• ManticoreQueryCacheNearLimit (warning): Использование кэша запросов превышает 90% от настроенного максимума в течение 10 минут. Это увеличивает риск вытеснения и обновления кэша, потенциально замедляя запросы; рассмотрите возможность увеличения кэша или оптимизации шаблонов запросов.
• ManticoreAgentRetryCountHigh (warning): manticore_agent_retry_count увеличивается более чем на 10 за 5 минут. Обычно указывает на проблемы с подключением к удаленным агентам (например, сетевые проблемы, таймауты или сбои в распределенных запросах).
• ManticoreCurrentConnectionsHigh (warning): Текущие подключения превышают 500 в течение 10 минут. Это может указывать на всплеск трафика, утечки подключений или медленных клиентов.
• ManticoreSlowestThreadHigh (warning): manticore_slowest_thread_seconds превышает 30 секунд в течение 10 минут. Присутствуют долго выполняющиеся или зависшие запросы; исследуйте медленные запросы и конкуренцию за ресурсы.
• ManticoreConnectTimeHigh (warning): manticore_connect_time_seconds превышает 0,2 секунды в течение 5 минут. Установка соединения задерживается (например, из-за сетевых проблем, накладных расходов TLS или перегрузки сервера).
• ManticoreSearchdCrashesRecently (critical): manticore_searchd_crashes_total увеличивается за последние 10 минут. Произошел сбой; проверьте логи, дампы памяти и события OOM.
• ManticoreBinlogFilesHigh (warning): Количество файлов бинарного лога превышает 1000 в течение 10 минут. Это говорит о том, что бинарные логи не ротируются или не очищаются должным образом; проверьте настройки бинарного лога и использование диска.
• ManticoreSearchdFdCountHigh (warning): Количество файловых дескрипторов searchd превышает 4096 в течение 10 минут. Это грозит достижением ограничений ОС; проверьте ulimit -n, подключения, открытые файлы и возможные утечки.
• ManticoreBuddyFdCountHigh (warning): Количество файловых дескрипторов Buddy превышает 4096 в течение 10 минут. Аналогично предыдущему, но для процесса Buddy.
• ManticoreSearchdAnonRssHigh (warning): Анонимная RSS searchd превышает 8 ГБ в течение 10 минут. Высокое использование памяти, не связанной с файлами; изучите рост памяти, шаблоны запросов/индексов и ограничения контейнера.
• ManticoreBuddyAnonRssHigh (warning): Анонимная RSS Buddy превышает 8 ГБ в течение 10 минут. Повышенное использование памяти Buddy; проверьте логи Buddy и рабочую нагрузку.
• ManticoreNonServedTables (warning): Одна или несколько таблиц, перечисленных в manticore.json, отсутствуют в SHOW TABLES в течение 10 минут. Обычно это означает, что таблицы не загрузились или были удалены/перемещены; проверьте логи запуска и пути к таблицам.
• ManticoreDiskMappedCacheLow (warning): Коэффициент дискового отображенного кэша падает ниже 50% в течение 15 минут. Вероятно увеличение дискового ввода-вывода, что потенциально замедляет поиск, если рабочий набор превышает оперативную память.
• ManticoreDiskMappedCacheVeryLow (critical): Коэффициент дискового отображенного кэша падает ниже 20% в течение 15 минут. Серьезные промахи кэша; ожидайте интенсивного дискового ввода-вывода и высокой задержки.

Дашборд организует панели по темам. Ключевые моменты включают:

• Здоровье / Время работы: Время с последнего перезапуска; частые падения указывают на перезапуски или сбои.
• Производительность / Задержка поиска (P95/P99): Метрики хвостовой задержки; постоянные увеличения часто сигнализируют о перегрузке или медленных запросах.
• Производительность / Длина очереди задач и Нагрузка / Давление в очереди: Индикаторы накопления и очереди; растущие очереди предполагают, что система отстает.
• Утилизация рабочих потоков % и Нагрузка / Активные/Всего рабочих потоков: Уровень загрузки рабочих потоков; значения, близкие к 100%, обычно коррелируют с более высокой задержкой.
• Производительность / Использование кэша запросов и Производительность / Коэффициент попаданий в кэш запросов: Нагрузка и эффективность кэша; высокое использование при низком коэффициенте попаданий может указывать на неэффективность.
• Таблицы / Коэффициент дискового отображенного кэша (Худшие 10): Низкие коэффициенты подразумевают больше дисковых чтений и потенциально более медленные запросы.

Этот подход использует коннектор MySQL Grafana для прямого запроса таблиц Manticore, позволяя визуализировать ваши поисковые данные, такие как временные тренды или агрегации.

Перед настройкой визуализации данных:

1. Убедитесь, что Manticore Search (версия 6.2.0 или новее) установлена и настроена. Подробности см. в официальном руководстве по установке Manticore Search.
2. Установите Grafana на вашу систему. Обратитесь к официальному руководству по установке Grafana для получения инструкций.

Чтобы подключить Manticore Search к Grafana:

1. Войдите в свою панель управления Grafana и нажмите на «Configuration» (иконка шестерёнки) в левой боковой панели.
2. Выберите «Data Sources» и нажмите «Add data source».
3. Выберите «MySQL» из списка доступных источников данных.
4. На странице настроек укажите следующие данные:
   • Name: Название источника данных (например, «Manticore Search»)
   • Host: Имя хоста или IP адрес сервера Manticore Search (с портом MySQL, по умолчанию: localhost:9306)
   • Database: Оставьте пустым или укажите имя вашей базы данных
   • User: Имя пользователя, имеющего доступ к Manticore Search (по умолчанию: root)
   • Password: Пароль для указанного пользователя (по умолчанию: пустой)
5. Нажмите «Save & Test», чтобы проверить подключение.

После подключения Manticore Search к Grafana вы можете создавать панели и визуализации:

1. В панели Grafana нажмите на значок «+» в левой боковой панели и выберите «New dashboard».
2. Нажмите кнопку «Add visualization», чтобы начать настройку графика.
3. Выберите источник данных Manticore Search, который вы подключили через MySQL-коннектор.
4. Выберите тип графика, который хотите создать (например, Time Series, Bar chart, Candlestick, Pie chart).
5. Используйте конструктор запросов Grafana или напишите SQL-запрос для получения данных из таблиц Manticore Search.
6. Настройте внешний вид графика, подписи и другие параметры по необходимости.
7. Нажмите «Apply», чтобы сохранить визуализацию на панели.

Вот простой пример с данными временных рядов. Сначала создайте таблицу и загрузите пример данных:

CREATE TABLE btc_usd_trading (
  id bigint,
  time timestamp,
  open float,
  high float,
  low float,
  close float
);

Загрузите данные:

curl -sSL https://gist.githubusercontent.com/donhardman/df109ba6c5e690f73198b95f3768e73f/raw/0fab3aee69d7007fad012f4e97f38901a64831fb/btc_usd_trading.sql | mysql -h0 -P9306

В Grafana вы можете создавать:

• График временных рядов: Визуализировать изменения цен со временем
• Свечный график: Отображать значения открытия, максимума, минимума и закрытия для финансовых данных
• Агрегационные графики: Использовать функции COUNT, AVG, MAX, MIN

Примеры запросов:

-- Time series query
SELECT time, close FROM btc_usd_trading ORDER BY time;

-- Aggregation query
SELECT DATE(time) as date, AVG(close) as avg_price
FROM btc_usd_trading
GROUP BY date
ORDER BY date;

При работе с Manticore Search через Grafana вы можете:

• Выполнять SQL-запросы к таблицам Manticore Search
• Использовать агрегатные функции: COUNT, AVG, MAX, MIN
• Применять операции GROUP BY и ORDER BY
• Фильтровать данные с помощью WHERE
• Получать метаданные таблиц через information_schema.tables
• Создавать различные типы визуализаций, поддерживаемые Grafana

• Некоторые продвинутые функции MySQL могут быть недоступны при работе с Manticore Search через Grafana.
• Доступны только функции, поддерживаемые Manticore Search. Подробнее смотрите в справочнике по SQL.

Для получения дополнительной информации и подробных инструкций:

• Пост в блоге об интеграции с Grafana
• Официальная документация Grafana

Kibana — это визуальный интерфейс, который позволяет исследовать, визуализировать и создавать панели управления для ваших логов. Интеграция Kibana с Manticore Search может ускорить загрузку визуализаций Kibana до 3 раз по сравнению с Elasticsearch, как продемонстрировано в этом демо. Эта интеграция позволяет пользователям беспрепятственно анализировать данные с помощью интерактивных панелей управления, пользовательских визуализаций и возможностей поиска в реальном времени. Она также упрощает работу с различными источниками данных, поддерживая инструменты, такие как Logstash и Filebeat для упрощённой загрузки данных, что делает её отличным выбором для рабочих процессов анализа логов.

1. Скачать Kibana: Убедитесь, что вы скачали версию Kibana, совместимую с Manticore. В настоящее время протестирована и рекомендована версия 7.6.0. Другие версии 7.x могут работать, но возможно возникновение проблем. Версия 8.x не поддерживается.
2. Проверить Manticore: Убедитесь, что ваш экземпляр Manticore работает и его HTTP API доступен (по умолчанию: http://localhost:9308).

1. Откройте файл конфигурации Kibana (kibana.yml).
2. Установите URL вашего экземпляра Manticore:

   elasticsearch.hosts: ["http://localhost:9308"]

3. Запустите Kibana и откройте её в браузере по адресу http://localhost:5601. При необходимости замените localhost на IP-адрес или имя хоста вашего сервера.

Примечание: Для работы с Kibana Manticore не требует настройки аутентификации. Также обратите внимание, что Manticore должен работать в режиме реального времени для интеграции с Kibana.

    listen = 127.0.0.1:9308:http
    pid_file = /var/run/manticore/searchd.pid
    data_dir = /var/lib/manticore
 }

## Supported Features
### Discover
- Use the **Discover** tab in Kibana to search and filter your data interactively.
- Refine your searches using the query bar with simple queries in the [Kibana query language](https://www.elastic.co/guide/en/kibana/current/kuery-query.html).

### Visualizations
- Navigate to **Visualizations** to create custom visualizations:
  - Create a table pattern (it’s called an 'index pattern' in Kibana) if one doesn’t already exist to define your data source.
  - Choose a visualization type (e.g., bar chart, line chart, or pie chart).
  - Configure your visualization, execute it, and explore your data.
  - Save your visualizations for future use.

### Dashboards
- Access **Dashboards** to create or view interactive dashboards:
  - Add visualizations, filters, or controls for a personalized experience.
  - Interact with your data directly from the dashboard.
  - Save dashboards for future use.

### Management
- Go to **Management > Kibana** to customize settings like default time zones and visualization preferences.

## Limitations
- Currently, Kibana version 7.6.0 is tested and recommended. Other 7.x versions may work but could cause issues. Versions 8.x are not supported.
- The following Elasticsearch-specific field types are not supported:
  - [Spatial data types](https://www.elastic.co/guide/en/elasticsearch/reference/current/mapping-types.html#spatial_datatypes)
  - [Structured data types](https://www.elastic.co/guide/en/elasticsearch/reference/current/mapping-types.html#structured-data-types)
  - [Document ranking types](https://www.elastic.co/guide/en/elasticsearch/reference/current/mapping-types.html#document-ranking-types)
  - [Text search types](https://www.elastic.co/guide/en/elasticsearch/reference/current/mapping-types.html#text-search-types) (except for plain 'text')
  - [Relational data types](https://www.elastic.co/guide/en/elasticsearch/reference/current/mapping-types.html#object-types)
- Metric aggregation functions are limited to [those supported by Manticore](../Searching/Grouping.md#Aggregation-functions).
- The following Kibana tools are not supported:
  - [Canvas](https://www.elastic.co/guide/en/kibana/7.6/canvas.html) – A visualization and presentation tool for combining data with colors and images.
  - [Elastic Maps](https://www.elastic.co/guide/en/kibana/7.6/maps.html) – A tool for analyzing geographical data.
  - [Metrics](https://www.elastic.co/guide/en/kibana/7.6/xpack-infra.html) – An app for monitoring infrastructure metrics.
  - [Logs](https://www.elastic.co/guide/en/kibana/7.6/xpack-logs.html) – A console-like display for exploring logs from common services.
  - Monitoring:
    - [Uptime](https://www.elastic.co/guide/en/kibana/7.6/xpack-uptime.html) – Monitors the status of network endpoints via HTTP/S, TCP, and ICMP.
    - [APM (Application Performance Monitoring)](https://www.elastic.co/guide/en/kibana/7.6/xpack-apm.html) – Collects in-depth performance metrics from applications.
    - [SIEM (Security Information and Event Management)](https://www.elastic.co/guide/en/kibana/7.6/xpack-siem.html) – An interactive workspace for security teams to triage events and conduct initial investigations.
    - [ILM (Index lifecycle management)](https://www.elastic.co/guide/en/elasticsearch/reference/7.6/index-lifecycle-management.html) - Automatically manage indices according to performance, resiliency, and retention requirements.
    - [Stack Monitoring](https://www.elastic.co/guide/en/kibana/7.6/xpack-monitoring.html) – Provides visualizations of monitoring data across the Elastic Stack.
  - [Elasticsearch Management](https://www.elastic.co/guide/en/kibana/7.6/management.html) – A UI for managing Elastic Stack objects, including ILM (Index Lifecycle Management), etc.

## Data Ingestion and Exploration
Integrate Manticore with tools like [Logstash](../Integration/Logstash.md), [Filebeat](../Integration/Filebeat.md), [Fluentbit](https://manticoresearch.com/blog/integration-of-manticore-with-fluentbit/), or [Vector.dev](https://manticoresearch.com/blog/integration-of-manticore-with-vectordev/) to ingest data from sources like web logs. Once the data is loaded into Manticore, you can explore and visualize it in Kibana.

ПРИМЕЧАНИЕ: эта функциональность требует Manticore Buddy. Если она не работает, убедитесь, что Buddy установлен.

Manticore поддерживает интеграцию с Apache Kafka для приема данных в реальном времени через источники Kafka и материализованные представления, что позволяет индексировать и искать данные в реальном времени. В настоящее время протестированы и поддерживаются apache/kafka версии 3.7.0-4.1.0.

Чтобы начать, вам необходимо:

1. Определить источник: Указать топик Kafka, из которого Manticore Search будет читать сообщения. Эта настройка включает такие детали, как хост брокера, порт и имя топика.
2. Настроить целевую таблицу: Выбрать таблицу Manticore реального времени для хранения входящих данных Kafka.
3. Создать материализованное представление: Настроить материализованное представление (mv) для обработки преобразования данных и сопоставления из Kafka в целевую таблицу в Manticore Search. Здесь вы определите сопоставления полей, преобразования данных, а также любые фильтры или условия для входящего потока данных.

CREATE SOURCE <source name> [(column type, ...)] [source_options]

allowed_field_name 'original JSON key name with special symbols' type

price_field '$price' float    -- maps JSON key '$price' to field 'price_field'
field_123 '123field' text     -- maps JSON key '123field' to field 'field_123'

CREATE SOURCE kafka
(id bigint, term text, abbrev '$abbrev' text, GlossDef json)
type='kafka'
broker_list='kafka:9092'
topic_list='my-data'
consumer_group='manticore'
num_consumers='2'
batch=50

Query OK, 2 rows affected (0.02 sec)

CREATE TABLE destination_kafka
(id bigint, name text, short_name text, received_at text, size multi);

Query OK, 0 rows affected (0.02 sec)

CREATE MATERIALIZED VIEW <materialized view name>
TO <destination table name> AS
SELECT [column|function [as <new name>], ...] FROM <source name>

CREATE MATERIALIZED VIEW view_table
TO destination_kafka AS
SELECT id, term as name, abbrev as short_name,
       UTC_TIMESTAMP() as received_at, GlossDef.size as size FROM kafka

Query OK, 2 rows affected (0.02 sec)

Данные передаются из Kafka в Manticore Search пакетами, которые очищаются после каждого запуска. Для вычислений между пакетами, таких как AVG, будьте осторожны, так как они могут работать не так, как ожидается, из-за обработки пакет за пакетом.

Вот таблица сопоставления на основе приведенных выше примеров:

SHOW SOURCES

+-------+
| name  |
+-------+
| kafka |
+-------+

SHOW SOURCE kafka;

+--------+-------------------------------------------------------------------+
| Source | Create Table                                                      |
+--------+-------------------------------------------------------------------+
| kafka  | CREATE SOURCE kafka                                               |
|        | (id bigint, term text, abbrev '$abbrev' text, GlossDef json)      |
|        | type='kafka'                                                      |
|        | broker_list='kafka:9092'                                          |
|        | topic_list='my-data'                                              |
|        | consumer_group='manticore'                                        |
|        | num_consumers='2'                                                 |
|        | batch=50                                                          |
+--------+-------------------------------------------------------------------+

SHOW MVS

+------------+
| name       |
+------------+
| view_table |
+------------+

SHOW MV view_table

+------------+--------------------------------------------------------------------------------------------------------+-----------+
| View       | Create Table                                                                                           | suspended |
+------------+--------------------------------------------------------------------------------------------------------+-----------+
| view_table | CREATE MATERIALIZED VIEW view_table TO destination_kafka AS                                            | 0         |
|            | SELECT id, term as name, abbrev as short_name, UTC_TIMESTAMP() as received_at, GlossDef.size as size   |           |
|            | FROM kafka                                                                                             |           |
+------------+--------------------------------------------------------------------------------------------------------+-----------+

ALTER MATERIALIZED VIEW view_table suspended=1

Query OK (0.02 sec)

Вы также можете указать partition_list для каждой темы Kafka. Одним из основных преимуществ такого подхода является возможность реализации шардинга вашей таблицы через Kafka. Для этого следует создать отдельную цепочку source → materialized view → destination table для каждого шарда:

Источники:

CREATE SOURCE kafka_p1 (id bigint, term text)
  type='kafka' broker_list='kafka:9092' topic_list='my-data'
  consumer_group='manticore' num_consumers='1' partition_list='0' batch=50;

CREATE SOURCE kafka_p2 (id bigint, term text)
  type='kafka' broker_list='kafka:9092' topic_list='my-data'
  consumer_group='manticore' num_consumers='1' partition_list='1' batch=50;

Таблицы назначения:

CREATE TABLE destination_shard_1 (id bigint, name text);
CREATE TABLE destination_shard_2 (id bigint, name text);

Материализованные представления:

CREATE MATERIALIZED VIEW mv_1 TO destination_shard_1 AS SELECT id, term AS name FROM kafka_p1;
CREATE MATERIALIZED VIEW mv_2 TO destination_shard_2 AS SELECT id, term AS name FROM kafka_p2;

• В этой конфигурации ребалансировка должна управляться вручную.
• По умолчанию Kafka не распределяет сообщения по схеме round-robin.
• Чтобы добиться распределения, похожего на round-robin при отправке данных, убедитесь, что ваш Kafka producer настроен с:
  • parse.key=true
  • key.separator={your_delimiter}

Иначе Kafka будет распределять сообщения по своим внутренним правилам, что может привести к неравномерному разделению по партициям.

Фиксация смещений Kafka происходит после каждой партии или при истечении времени обработки. Если процесс неожиданно остановится во время выполнения запроса материализованного представления, могут появиться дубликаты записей. Чтобы избежать этого, включите в схему поле id, позволяющее Manticore Search предотвращать дублирование в таблице.

• Инициализация работника: После настройки источника и материализованного представления Manticore Search создает выделенного работника для обработки поступления данных из Kafka.
• Отображение сообщений: Сообщения сопоставляются согласно схеме конфигурации источника, преобразуясь в структурированный формат.
• Группировка в партии: Сообщения группируются в партии для эффективной обработки. Размер партии можно настроить в зависимости от ваших требований к производительности и задержкам.
• Буферизация: Партии преобразованных данных сохраняются во вспомогательной таблице буфера для эффективных массовых операций.
• Обработка материализованного представления: К данным во вспомогательной таблице применяется логика представления, выполняются преобразования или фильтрация.
• Передача данных: Обработанные данные затем передаются в конечную таблицу реального времени.
• Очистка: Вспомогательная таблица очищается после каждой партии, чтобы подготовиться к приему следующего набора данных.