Основные функции MatrixGate

В этом документе описываются основные функции MatrixGate.

Примечание!
Для доступа к MatrixGate из языков программирования см. Загрузка данных — Доступ к MatrixGate из языков программирования.

1 Загрузка специальных типов данных через MatrixGate

1.1 Пример загрузки CSV-файлов через MatrixGate

Создайте таблицу csvtable в базе данных demo.

demo=# CREATE TABLE csvtable (
     time TIMESTAMP WITH TIME ZONE,
     tagid INT,
     c1 INT,
     c2 INT,
     c3 INT
     )
     USING MARS3
     DISTRIBUTED BY (tagid)
     ORDER BY (time,tagid);

Отредактируйте файл загрузки данных data.csv, содержимое следующее:
```
1603777821|1|101|201|301
1603777822|2|102|202|302
1603777823|3|103|203|303
```
Запустите mxgate, укажите параметр --source со значением stdin, целевую таблицу — существующую csvtable, степень параллелизма загрузки — 2. В примере хост — mdw.
```
[mxadmin@mdw ~]$ mxgate \
--source stdin \
--db-database demo \
--db-master-host 127.0.0.1 \
--db-master-port 5432 \
--db-user mxadmin \
--time-format unix-second \
--delimiter "|" \
--target csvtable \
--parallel 2 < data.csv
```

Подключитесь к базе данных и проверьте успешность загрузки данных.


demo=# SELECT * FROM csvtable ;
        time          | tagid | c1  | c2  | c3
-----------------------+-------+--------------------------------------------------------------------------------------------------
2020-10-27 05:50:23+08 |     3 | 103 | 203 | 303
2020-10-27 05:50:22+08 |     2 | 102 | 202 | 302
2020-10-27 05:50:21+08 |     1 | 101 | 201 | 301

(3 rows)

### 1.2 Пример загрузки JSON-полей через MatrixGate

#### 1.2.1 JSON

- Создайте таблицу.

demo=# CREATE TABLE json_test( id int, j json ) USING MARS3 ORDER BY (id);

- Создайте файлы данных.
 `~/json.csv`

1|"{""a"":10, ""b"":""xyz""}"

- Загрузка
Здесь используется режим stdin, другие режимы аналогичны.
Ключевой параметр — `--format csv`.

[mxadmin@mdw ~]$ mxgate \ --source stdin \ --db-database postgres \ --db-master-host 127.0.0.1 \ --db-master-port 7000 \ --db-user mxadmin \ --time-format raw \ --format csv \ --delimiter "|" \ --target json_test < ~/json.csv

- Просмотрите загруженные данные.

demo=# SELECT * FROM json_test; id | j ----+----------------------------------------------------------------------------------------------------------------------------- 1 | {"a":10, "b":"xyz"} (1 row)

#### 1.2.2 Массив JSON

- Создайте таблицу.

demo=# CREATE TABLE json_array_test( id int, j json ) USING MARS3 ORDER BY (id);

- Создайте файлы данных
 `~/json_array.csv`.

1|"{""{\""a\"":10, \""b\"":\""xyz\""}"",""{\""c\"": 10}""}"

- Загрузите через mxgate.

- Проверьте результат.

demo=# SELECT * FROM json_array_test ; id | j ----+----------------------------------------------------------------------------------------------------------------------------- 1 | {"{\"a\":10, \"b\":\"xyz\"}","{\"c\": 10}"} (1 row)

> ***Примечание!***  
Поскольку столбец JSON содержит специальные символы, такие как кавычки, параметр --format для mxgate должен быть установлен в CSV.

<a name="watch"><br/></a>
## 2 Наблюдение за показателями работы mxgate

`watch` — это подкоманда mxgate, использующая ряд показателей для описания работы демона mxgate.
Существует два режима `watch`:
- Режим реального времени: вывод метрик шлюза каждые 3 секунды в формате, аналогичном sar.
- Режим исторических наблюдений: можно указать любой временной период (например, каждый час вчера, каждый день прошлого месяца, каждый месяц прошлого года) для статистики скорости импорта.

### 2.1 Наблюдение в реальном времени

[mxadmin@mdw ~]$ mxgate watch

Показатели работы mxgate будут собираться каждые три секунды, результат вывода следующий:

             Time          WCount          ICount        WSpeed/s        ISpeed/s  WBandWidth MB/S     BlocakItems

2022-04-28 15:20:58 14478858 14527011 2598081 2627887 2395 0 2022-04-28 15:21:01 22231035 22633254 2584059 2702081 2222 0 2022-04-28 15:21:04 30494310 30500874 2754425 2622540 3551 0 2022-04-28 15:21:07 38004210 38032956 2503300 2510694 2862 0 2022-04-28 15:21:10 46188696 46298223 2728162 2755089 2227 0 ...

Описание каждого показателя можно получить с помощью параметра `--info`

[mxadmin@mdw ~]$ mxgate watch --info

По умолчанию выводятся только показатели скорости, а временные показатели можно наблюдать с помощью параметра --watch-latency для анализа проблем.

[mxadmin@mdw ~]$ mxgate watch --watch-latency

### 2.2 Наблюдение за историческими данными

[mxadmin@mdw ~]$ mxgate watch --history

Будет рассчитана средняя скорость за каждый час в течение последних 24 часов на текущий момент, результат вывода следующий:

            TIME RANGE                | SPEED/S  | BANDWIDTH MB/S  | BLOCK ITEMS

2022-04-28 16:00:00-2022-04-28 17:00:00 | 2208010 | 1254.48 | 0 2022-04-28 17:00:00-2022-04-28 18:00:00 | 1157920 | 1327.00 | 0 2022-04-28 18:00:00-2022-04-28 19:00:00 | 2228666 | 2162.32 | 0 2022-04-28 19:00:00-2022-04-28 20:00:00 | 1371092 | 2881.30 | 0 2022-04-28 20:00:00-2022-04-28 21:00:00 | 1575320 | 2608.20 | 0

SPEED/S, BANDWIDTH MB/S обозначают скорость импорта записей и пропускную способность импорта (в МБ/с),
BLOCK ITEMS — объем данных, заблокированных в mxgate. Это значение возрастает, когда скорость потребления данных в базе не успевает за скоростью генерации данных источниками (http, kafka и т.д.).

Можно добавить параметры `--watch-start`, `--watch-end`, `--watch-duration` для управления интервалом и периодом наблюдения за историческими данными.
Например:

[mxadmin@mdw ~]$ mxgate watch --history --watch-start '2022-03-27 00:00:00' --watch-end '2022-04-27 00:00:00' --watch-duration '168h'

Средняя скорость импорта по неделям (каждые 168 ч) с 27 марта по 27 апреля.
Параметр `--watch-duration` поддерживает три единицы измерения: `h`` `m`` `s`

<a name="non-stop_update_parameters"><br/></a>
## 3 Обновление параметров параллельной записи без остановки

mxgate поддерживает изменение параметров параллельной загрузки без остановки: `--interval`, `--stream-prepared`. Параметр `--interval` определяет рабочее время каждого соединения записи от mxgate к таблице базы данных, а `--stream-prepared` — количество активных соединений записи. Согласно логике mxgate, для одной и той же таблицы базы данных одновременно работает только одно соединение записи. Поэтому для каждой таблицы требуется несколько соединений, чтобы выполнять задачи записи в разные временные интервалы, обеспечивая высокую скорость и эффективность записи. В этом процессе вы можете регулировать рабочее время каждого соединения записи с помощью параметра `--interval`, тем самым целенаправленно повышая скорость записи данных и производительность загрузки. Конкретный пример использования:

- Установите количество соединений записи на таблицу равным 3

$ mxgate set --stream-prepared-cli 3

| \/ | | | _ () / _| | | __ | |\/| |/ ` | | '| \ \/ / | / ` | / \ | | | | (| | || | | |> <| || | (_| | || / || ||_,|_|| |//_\|\,_|\_| Version: v5.1.0 Your Copy is Licensed to: yMatrix.cn; 2023-10-26; any

- Get the number of active write connections per table

$ mxgate get --stream-prepared-get

public.vehicle_basic_data_mars3 stream-prepared = 3

- Set the time interval for write connections of all tables to 200ms

$ mxgate set --job-interval 200

- Get the current time interval for write connections to all tables

$ mxgate get --job-interval-get

> ***Примечание!***  
Для указанных выше параметров, если необходимо установить или получить количество соединений записи или рабочее время для конкретной таблицы, добавьте после команды параметр `--job <name>`. Каждая задача соответствует одной таблице базы данных. Структура параметра job состоит из имени схемы и имени таблицы. Например, если ваша таблица называется test_table, а схема — public, нужно добавить `--job public.test_table` после существующей команды.

<a name="non-stop_update_table"><br/></a>
## 4 Изменение структуры таблицы без остановки

В процессе загрузки данных может возникнуть необходимость изменить структуру таблицы, поскольку источники временных рядов изменились, и ранее заданная структура больше не подходит. Такая потребность может быть удовлетворена с помощью mxgate. В этом разделе объясняется, как mxgate выполняет серию операций без остановки: приостановка записи данных, перезагрузка изменённой метаинформации таблицы базы данных и восстановление записи данных. Конкретные шаги следующие:

 * Во-первых, используйте команду `mxgate pause -X`, чтобы прервать соединения записи всех таблиц, подготовившись к изменению структуры таблицы базы данных. Параметр `-X` является обязательным — он помогает прервать соединение между mxgate и базой данных. Если соединение не будет прервано, изменить таблицу базы данных невозможно. Кроме того, используйте параметр `-S`, чтобы команда pause синхронно дождалась завершения прерывания всех соединений перед возвратом.

$ mxgate pause -X

begin to pause all jobs，please wait...

* Во-вторых, после прерывания всех соединений записи с соответствующей таблицей можно изменить структуру таблицы базы данных, например добавить столбцы, удалить столбцы, удалить существующую таблицу и создать новую таблицу с тем же именем.

> ***Примечание!***  
Структура пересозданной таблицы может отличаться, но «имя таблицы» должно оставаться одинаковым.

 * Наконец, используйте команду `mxgate resume -R`, чтобы восстановить соединения записи для всех таблиц и перезагрузить метаинформацию таблицы данных. Параметр `-R` обязателен, `resume` и `-R` совместно выполняют операцию перезагрузки.

$ mxgate resume -R

begin to reload all jobs，please wait...

* Особо важно: когда одновременно запущено несколько процессов mxgate, необходимо использовать параметр `-p`, чтобы указать номер процесса соответствующего mxgate. Все вышеперечисленные команды аналогичны, например:

$ mxgate get --job-interval-get -p 70199 --job

> ***Примечание!***  
Условие выполнения команды перезагрузки — все соединения записи соответствующей таблицы mxgate должны быть предварительно приостановлены, иначе произойдет ошибка: `jobs are not paused, please pause them first`

<a name="upsert"><br/></a>
## 5 Поддержка функции UPSERT

YMatrix реализует функцию UPSERT в MatrixGate начиная с версии v4.2.0. Подробности см. в разделе [Сценарий пакетного слияния данных (UPSERT)](/ru/doc/5.1/datainput/upsert).

<a name="transfer"><br/></a>
## 6 Миграция одной таблицы

YMatrix реализует функцию миграции одной таблицы в MatrixGate начиная с версии v4.3.0. Подробности см. в разделе [Миграция одной таблицы](/ru/doc/5.1/maintain/migrate/mxgate).

>***Примечание!***  
YMatrix поддерживает не только миграцию одной таблицы, но и миграцию всей базы данных (v4.7.0 и выше). Подробности см. в разделе [Миграция всей базы данных](/ru/doc/5.1/maintain/migrate/mxshift).

 <a name="log_level"><br/></a>
## 7 Динамическое изменение уровня логирования без остановки

Иногда нам нужно включить отладочный лог mxgate для наблюдения за некоторыми ключевыми информационными сообщениями, но включение или отключение отладочного лога требует перезапуска mxgate, что неудобно для диагностики проблем. Поэтому YMatrix предоставляет возможность динамически изменять уровень логирования mxgate:

 * Когда gate работает, используйте команду `mxgate set --log-level VERBOSE`, чтобы включить лог уровня `VERBOSE` с относительно полной информацией, или `mxgate set --log-level DEBUG`, чтобы включить лог уровня `DEBUG` с максимально полной информацией. Когда отладочный лог больше не нужен, используйте `mxgate set --log-level INFO`, чтобы вернуть уровень логирования к `INFO`.

<a name="eventtrigger"><br/></a>
 ## 8 Событийные триггеры mxgate

Иногда во время операции вставки через mxgate требуется выполнить некоторые DDL-изменения в таблице. Эта функция позволяет mxgate «приостановить» входящие операции вставки после получения инструкции DDL-изменения, чтобы DDL-оператор мог быть выполнен как можно быстрее.

Поддерживаемые DDL-операции:
 1. Очистка таблицы или группы таблиц

=# TRUNCATE TABLE ;

2. Добавление столбцов, удаление столбцов и изменение типов столбцов в таблице

=# ALTER TABLE ADD COLUMN ; =# ALTER TABLE DROP COLUMN ; =# ALTER TABLE ALTER COLUMN TYPE ;

3. Изменение имени таблицы

=# ALTER TABLE RENAME TO ;

4. Удаление таблицы с последующим её повторным созданием

=# DROP TABLE ; =# CREATE TABLE <tablename(column datatype)>;

5. Вышеуказанные четыре DDL-операции для партиционированных таблиц

<a name="autotune"><br/></a>
## 9 Автоматическая настройка количества соединений в mxgate

Обычно мы можем адаптировать нагрузку (то есть объем данных для записи) различных задач записи (Jobs), вручную регулируя [параметр stream-prepared](/ru/doc/5.1/tools/mxgate/parameter#streamprepared) (количество параллельно выполняемых сессий соединения). Однако в некоторых бизнес-сценариях нагрузка на задачи записи постоянно меняется динамически, и эти изменения невозможно корректировать вручную.

Поэтому MatrixGate разработал автоматический механизм: автоматическая адаптация нагрузки записи осуществляется путем включения [параметра auto-tune](/ru/doc/5.1/tools/mxgate/parameter#autotune).

Этот параметр можно включить следующими способами:
- Передача параметра `--auto-tune` при запуске mxgate
- Настройка в конфигурационном файле mxgate
- Динамическое управление с помощью команды `mxgate set`
   - Включить: `mxgate set --auto-tune`
   - Выключить: `mxgate set --auto-tune=false`

>***Примечание!***  
Включение этой функции гарантирует максимально эффективную производительность записи в реальном времени, но не обязательно оптимально с точки зрения использования ресурсов сессионных соединений.

>***Примечание!***  
Если вы включили автоматическую настройку соединений и установили [порог высокой нагрузки] (#sourcepressure), функция автоматической настройки соединений будет работать только при превышении порога нагрузки записи.

<a name="sourcepressure"><br/></a>
## 10 mxgate Процентная квантилизация нагрузки на запись

mxgate поддерживает квантификацию нагрузки на запись исходных данных (в процентах от 0 до 100%). Во время процесса записи данных через mxgate, нагрузку на запись соответствующей задачи (Job) можно получить с помощью следующей команды:

$ mxgate get --source-pressure

public.t2 source-pressure = 27.26%

If there are multiple jobs, you can get the corresponding write load through the name of the job, or get the write load of multiple jobs at once:

$ mxgate get --source-pressure --job public.t2

public.t2 source-pressure = 15.73%

$ mxgate get --source-pressure

public.t2 source-pressure = 35.84% public.t1 source-pressure = 0%

After completing the percentage load quantization, you can use this metric to set a corresponding high water level threshold. If the data writing load at the source end exceeds the set threshold, the corresponding number will be displayed as **red**. At this time, you can manually adjust the number of connections to reduce the write pressure. Of course, you can also use the [mxgate automatic adjustment of connections] (#autotune) function.

The threshold value needs to be set through the following command:

$ mxgate set --high-water-mark --job

For example:

$ mxgate set --high-water-mark 20 --job public.t2

public.t2 high-water-mark = 20.00%

Get the high water level threshold of the corresponding job through the following command:

$ mxgate get --high-water-mark-get --job public.t2

public.t2 high-water-mark = 20.00%

$ mxgate get --high-water-mark-get

public.t1 high-water-mark = 0.00% public.t2 high-water-mark = 20.00%

>***Примечание!***  
Если установленный порог превышен, значение давления отобразится красным цветом.

Если вы включите функцию автоматической настройки количества соединений в mxgate и установите соответствующий порог высокого уровня, количество соединений не будет автоматически увеличиваться, пока нагрузочное давление задачи не опустится ниже установленного порога высокого уровня. В этом случае попытки увеличения количества соединений через функцию автоматической настройки будут бесполезны.

Например: Мы устанавливаем порог высокого уровня `75%`. Когда нагрузочное давление составляет `35%`, хотя mxgate определяет, что количество соединений следует увеличить, из-за того, что `75%` установленного порога высокого уровня ещё не достигнуто, количество соединений в этот момент не увеличится, а будет выведено следующее сообщение об ошибке:

[WARN]:-[SlotsLauncher] Sources pressure = 35.21 of job(public.t2) is lower than the high water mark(75), the slot increase is not allowed

>***Примечание!***  
Приведённые значения являются примерными. Устанавливайте соответствующий порог высокого уровня в зависимости от реальных бизнес-требований и текущей скорости записи данных.

Если вы хотите снять ограничение по порогу высокого уровня, используйте следующую команду:

$ mxgate set --disable-high-water-mark=true --job public.t2

public.t2 high-water-mark = 0.00%

История версий

Русский English 简体中文

Основные функции MatrixGate

1 Загрузка специальных типов данных через MatrixGate

1.1 Пример загрузки CSV-файлов через MatrixGate

Русский

English

简体中文