Настройка ETL-процесса

В рамках системы вы можете настроить конвейер коллектора для проведения ETL-процесса (Extract, Transfer, Load) в целях решения определенных задач бизнес-аналитики.

Настройка ETL-процесса состоит из следующих этапов:

Настройка БД PostgreSQL
Предварительная настройка системы
Настройка конвейера ETL

Настройка БД PostgreSQL

Чтобы подготовить БД PostgreSQL к ETL-процессу:

Подключитесь к серверу, на котором развернута БД PostgreSQL.
Установите плагин WAL2json для PostgreSQL.
Откройте файл postgresql.conf.
Вы можете узнать расположение файла postgresql.conf, выполнив SQL-запрос:

SHOW config_file
Добавьте или измените следующие параметры в файле postgresql.conf:
```
wal_level=logical
max_replication_slots=2
max_wal_senders=10
max_slot_wal_keep_size=1024
```

Перезапустите БД PostgreSQL для применения настроек.

Пример команды перезапуска, если БД PostgreSQL установлена как системная служба на сервере с ОС Linux:

sudo systemctl restart <postgresql>

Здесь:

<postgresql> — название сервиса postgresql, например, postgresql или postgresql-16.

Подключитесь к БД PostgreSQL, выполнив команду:
```
psql -U postgres -d auth
```
Создайте логический слот репликации, выполнив SQL-запрос:
```
SELECT * FROM pg_create_logical_replication_slot('replication_slot', 'wal2json');
```

Предварительная настройка системы

Чтобы подготовить систему к настройке ETL-процесса:

Зарегистрируйте необходимую базу данных событий для хранения данных в БД ClickHouse, если она еще не зарегистрирована.

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

objectId c типом UUID, чтобы различать объекты в хранилище событий.
isDeleted c типом UInt разрядности 8, чтобы исключить из отображения удаленные объекты.

Для корректной работы процесса идентификаторы полей в модели событий должны соответствовать идентификаторам полей в БД PostgreSQL.

Создайте хранилище событий. Выбор движка хранилища зависит от сценария ETL:

MergeTree — для исторических событий.

ReplacingMergeTree — для операционных событий.

Для данного движка требуется дополнительно задать следующие настройки:

На этапе Информация:
1. В поле ver выберите значение cnt, чтобы определять актуальную версию объекта по последнему событию.
2. В поле is_deleted выберите значение isDeleted, чтобы не отображать удаленные объекты.
3. Установите переключатель Параметры ключей в активное положение.

На этапе Параметры ключей добавьте сортировочный ключ objectId.

При сохранении изменений в блоке Параметры ключей без сортировки по полю timestamp отобразится предупреждение о нарушении порядка сортировки. В рамках сценария ETL-процесса для работы с операционными событиями предупреждение можно проигнорировать.

Рекомендуется под каждый метод проверки жизнеспособности идеи создавать отдельное хранилище.

Создайте секрет с данными для подключения к БД PostgreSQL, если он еще не создан.

Настройка конвейера ETL

Чтобы настроить конвейер с ETL-процессом:

Перейдите в раздел Ресурсы → Коллекторы.
Создайте коллектор для ETL-процесса, если он еще не создан.
Откройте карточку коллектора.
Перейдите на вкладку Конвейеры.
Добавьте конвейер для ETL-процесса.
Откройте конфигурацию конвейера.
Удостоверьтесь, что в выпадающем списке Версия на панели инструментов выбран вариант Черновик.
Добавьте точку входа со следующими параметрами:
1. Выберите тип точки входа Database.
2. Выберите актуальный драйвер БД PostgreSQL.
3. В поле Адрес подключения выберите секрет, созданный на этапе подготовки.
4. Введите следующий SQL-запрос для чтения изменений (CDC) из WAL-журнала:
  SELECT * FROM pg_logical_slot_get_changes('replication_slot', NULL, 100000, 'format-version', '2');
5. Остальные поля заполните в соответствии с конфигурацией системы.

Добавьте VRL-трансформацию для преобразования входных данных в формат отдельных событий, обеспечивающий их корректную обработку в Vector. Настройте VRL-трансформацию следующим образом:

Введите в поле VRL следующий код трансформации:

msg = parse_json!(.raw.data)
action = to_string(msg.action) ?? ""

saved_collector_id = .collectorId
saved_aggregation_rule_name = .aggregationRuleName
saved_aggregation_rule_id = .aggregationRuleId

if action == "I" || action == "U" || action == "D" {
    schema = to_string(msg.schema) ?? ""
    table = to_string(msg.table) ?? ""

    row = {}
    types = {}
    if action == "I" || action == "U" {
        for_each(array!(msg.columns)) -> |idx, col| {
            name = to_string(col.name) ?? ""
            row = set!(row, [name], col.value)
            types = set!(types, [name], to_string!(col.type) ?? "")
        }
    }

    keys = {}
    if action == "U" || action == "D" {
        if exists(msg.identity) {
            for_each(array!(msg.identity)) -> |idx, key| {
                key_name = to_string(key.name) ?? ""
                keys = set!(keys, [key_name], key.value)
            }
        }
    }

    object_id = ""
    if exists(row.id) {
        object_id = to_string(row.id) ?? ""
    } else if exists(keys.id) {
        object_id = to_string(keys.id) ?? ""
    }

    is_deleted = 0
    if action == "D" {
        is_deleted = 1
    }

    lsn_string = to_string(.raw.lsn) ?? "0/0"
    lsn_parts = split(lsn_string, "/") ?? ["0", "0"]
    lsn_hi = parse_int(to_string(lsn_parts[0]) ?? "0", base: 16) ?? 0
    lsn_lo = parse_int(to_string(lsn_parts[1]) ?? "0", base: 16) ?? 0
    cnt = lsn_hi * 4294967296 + lsn_lo

    . = {}

    .id = object_id
    .objectId = object_id
    .isDeleted = is_deleted
    .operation = action
    .timestamp = now()
    .eventTime = to_string(.timestamp) ?? ""
    .lsn = lsn_string
    .xid = to_string(.raw.xid) ?? ""
    .changeIndex = 0
    .sourceIp = "::"
    .tenantId = to_string(row.tenantId) ?? ""
    .organizationId = to_string(row.organizationId) ?? ""
    .raw = encode_json(msg) ?? "{}"

    .collectorId = to_string(saved_collector_id) ?? ""
    .aggregationRuleName = to_string(saved_aggregation_rule_name) ?? ""
    .aggregationRuleId = to_string(saved_aggregation_rule_id) ?? ""

    .cnt = cnt
    .type = 1
    .correlationRuleName = to_string(.correlationRuleName) ?? ""
    .correlationRuleId = to_string(.correlationRuleId) ?? ""
    .baseEventIds = array(.baseEventIds) ?? []
    .baseEventTimestamps = array(.baseEventTimestamps) ?? []
    .groupedBy = array(.groupedBy) ?? []
    .correlationSeverity = 1
    .routeSchema = schema
    .routeTable = table

    for_each(row) -> |key, value| {
        col_type = to_string!(get(types, [key]) ?? "")

        if contains(col_type, "timestamp") || contains(col_type, "date") {
            if is_string(value) && value != null && value != "" {
                value = parse_timestamp(value, format: "%Y-%m-%d %H:%M:%S%.f") ??
                        parse_timestamp(value, format: "%Y-%m-%d %H:%M:%S") ?? value
            }
        } else if contains(col_type, "jsonb") {
            if is_string(value) && value != null && value != "" {
                value = parse_json(value) ?? value
            }
        }

        . = set!(., [key], value)
    }
}

Остальные поля заполните в соответствии с конфигурацией системы.

Добавьте маршрутизатор для распределения потоков данных по конечным точкам. Для каждой планируемой конечной точки с хранилищем событий настройте свою маршрутизацию.
Добавьте конечные точки для распределения обработанных событий в хранилища событий. Настройте каждую конечную точку следующим образом:
1. Выберите тип конечной точки Хранилище событий.
2. Выберите хранилище событий, созданное на этапе подготовки.
3. Остальные поля заполните в соответствии с конфигурацией системы.
Рекомендуется настраивать отдельную конечную точку под каждый метод проверки жизнеспособности идеи.
Настройте связь между элементами конвейера:
1. Подключите точку входа к VRL-трансформации.
2. Подключите VRL-трансформацию к маршрутизатору.
3. Подключите маршруты маршрутизатора к соответствующим конечным точкам.
Включите коллектор.
Установите конфигурацию конвейера.

Была ли полезна эта страница?

Обратная связь