Утилита миграции данных из внешних источников

Запросы к ClickHouse

Запрос на получение метрик по объектам:

SELECT 
	sum(count) as count,
	multiIf(portal_id in &ПереходыВыходы AND metric_type == 'out', 'in', portal_id in &ПереходыВыходы AND metric_type == 'in', 'out', toString(metric_type)) as metric_type,
	toStartOfDay(toDateTime(timestamp, &ЧасовойПояс)) as timestampDay
FROM
	metrics_by_portals FINAL 
WHERE
	portal_id in  &ПереходыВсе
	AND toDateTime(timestamp, &ЧасовойПояс) BETWEEN toDateTime(&ДатаНачалаОтбора, &ЧасовойПояс) AND toDateTime(&ДатаОкончанияОтбора, &ЧасовойПояс)
	AND HOUR(toDateTime(timestamp, &ЧасовойПояс)) IN &МассивЧасовРаботы
	AND multiIf(HOUR(toDateTime(timestamp, &ЧасовойПояс)) = &ЧасНачалаРаботы, MINUTE(toDateTime(timestamp, &ЧасовойПояс)) >= &МинутаНачалаРаботы, HOUR(toDateTime(timestamp, &ЧасовойПояс)) = &ЧасОкончанияРаботы, MINUTE(toDateTime(timestamp, &ЧасовойПояс)) < &МинутаОкончанияРаботы, MINUTE(toDateTime(timestamp, &ЧасовойПояс)) >= 0)
GROUP BY
(
	metric_type,
	timestampDay
)
HAVING metric_type = 'in' 
SETTINGS optimize_move_to_prewhere_if_final = 1

где, 

Отбор по времени работы объекта использовать только в случае если объект не круглосуточный. 

 

Пример заполненного запроса:

SELECT 
	sum(count) as count,
	multiIf(portal_id in ('074d4fdd-6b5d-417b-a9b3-ee84b52b2252') AND metric_type == 'out', 'in', portal_id in ('4ad9d597-f21d-45dc-880a-72926630edea') AND metric_type == 'in', 'out', toString(metric_type)) as metric_type,
	toStartOfDay(toDateTime(timestamp, 'Asia/Yekaterinburg')) as timestampDay
FROM
	metrics_by_portals FINAL 
WHERE
	portal_id in ('05b6bef9-5c34-41b4-ae8f-8e09869e3f12', '074d4fdd-6b5d-417b-a9b3-ee84b52b2252')
	AND toDateTime(timestamp, 'Asia/Yekaterinburg') BETWEEN toDateTime('2023-03-26 00:00:00', 'Asia/Yekaterinburg') AND toDateTime('2023-03-26 23:59:59', 'Asia/Yekaterinburg')
	AND HOUR(toDateTime(timestamp, 'Asia/Yekaterinburg')) IN (5,6,7,8,9,10,11,12,13,14,15,16,17,18,19,20,21)
	AND multiIf(HOUR(toDateTime(timestamp, 'Asia/Yekaterinburg')) = 5, MINUTE(toDateTime(timestamp, 'Asia/Yekaterinburg')) >= 0, HOUR(toDateTime(timestamp, 'Asia/Yekaterinburg')) = 21, MINUTE(toDateTime(timestamp, 'Asia/Yekaterinburg')) < 25, MINUTE(toDateTime(timestamp, 'Asia/Yekaterinburg')) >= 0)
GROUP BY
(
	metric_type,
	timestampDay
)
HAVING metric_type = 'in' 
SETTINGS optimize_move_to_prewhere_if_final = 1

 

Получение данных по объекту:

SELECT 
	object_id as object_id,
	name as name, 
	portal_id as portal_id, 
	direction as direction,
	working_hours.timezone,
	working_hours.start_time,
	working_hours.end_time,
	working_hours.around_the_clock 
FROM 
	objects
	LEFT ANY JOIN working_hours AS wh
	  ON working_hours.object_id = objects.object_id
 ORDER BY object_id

Пример результата запроса:

image-1696854188264.png

Описание

Настройки

Настройка производится путем изменения значений переменных среды или ENV файла.

ENV файл представляется собой текстовый файл, в котором описаны настройки в формате КЛЮЧ=ЗНАЧЕНИЕ. Каждая строка файла может содержать одну пару КЛЮЧ=ЗНАЧЕНИЕ. Для ENV файла допускается имя .env и win.env. Значение переменных из файла ENV имеют более высокий приоритет чем переменные среды.

Описание настроек

ниже описаны настройки, разбитые на группы

Настройки приложения
Настройки локальной базы приложения
Настройки расписаний запуска задач
Настройки для взаимодействия с index-сервисом
Настройки адаптера CLICKHOUSE
Настройки адаптера COUNTMAX

возможно задание нескольких экземпляров.

номер экземпляра задается необязательным окончанием _NUM,
где NUM - целое положительное число

например
SPPINDEX_COUNTMAX_HOST - задает адрес сервера для нулевого экземпляра адаптера
SPPINDEX_COUNTMAX_HOST_1 - задает адрес сервера для первого экземпляра адаптера

сортировка адаптеров осуществляется по номеру экземпляра
допускается пропуск номеров

в описании переменных необязательное окончание обозначается квадратными скобками

Облако

Тестовый экземпляр утилиты миграции развернут в облаке. spp_agent_dev -> index-migration

Взаимодействие с index-сервисом

В зависимости от расписания заданного в SPPINDEX_CRON_UPLOAD запускается задача загрузки данных в index-сервис.

Загружаются только раннее не синхронизированные данные.

Последовательность действий итерации:

  1. Получение из локально кеша объектов по которым есть несихронизированные метрики
  2. Для каждого объекта из пункта 1 делается запрос http запрос в index-сервис для сохранения. В ответе получаем id объекта в базе сервиса
  3. Получение несинхронизированных метрик из локального кеша.
  4. Отправка данных из пункта 3 пачками в index-сервис
  5. При успешном запросе в п.4 помечаем метрики как синхронизированные

Описание API для взаимодействия с index-сервисом

Запрос сохранения объекта

POST: SPPINDEX_SERVICE_URL/control/clients/1/init_organization_object

{
    "external_id": OBJECT_ID
	"name": OBJECT_NAME
}

Ответ сервиса:

{
    "error": ERROR_FLAG,
    "message": MESSAGE,
    "id": REMOTE_OBJECT_ID
}
Запрос сохранения метрик

POST: SPPINDEX_SERVICE_URL/data/visits

[
  	...,
    {
  		"object_id": REMOTE_OBJECT_ID,
  		"day": DATE,
  		"visits": VISITS_COUNT
 	},
	...
]

Ответ сервиса:

{
    "error": ERROR_FLAG,
    "message": MESSAGE
}