2. Конфигурация CSV-uploader (application.yml)

Файл application.yml – основной конфигурационный файл модуля CSV-uploader, в котором задана логика и порядок работы загрузчика, а также другие настройки необходимые для корректной работы адаптера.

2.1. Пример файла application.yml

# Kafka Prostore
.kafkaUrl: &kafkaUrl ${KAFKA_BOOTSTRAP_SERVERS:localhost:9092}

http-server:
        # Порт для старта веб сервера
        port: ${HTTP_PORT:8080}
        # Включить веб-сервер
        enabled: ${HTTP_ENABLED:true}

send:
        # Размер отправляемой порции данных
        chunk-row-count: ${CHUNK_ROW_COUNT:1000}
        # Размер буфера на чтение файла
        file-buffer-size: ${FILE_BUFFER_SIZE:1048576}
        # Количество Job на чтение
        read-job-count: ${READ_JOB_COUNT:4}
        # Размер Channel для сериализации
        serialize-channel-size: ${SERIALIZE_CHANNEL_SIZE:20}
        # Количество Job на сериализацию
        serialize-job-count: ${SERIALIZE_JOB_COUNT:4}
        # Размер Job на отправку
        send-channel-size: ${SEND_CHANNEL_SIZE:20}
        # Количество Job на отправку
        send-job-count: ${SEND_JOB_COUNT:4}

file-size:
        # Ограничение на размер отправляемого файла (мегабайты)
        restriction: ${SEND_FILE_SIZE_RESTRICTION:1024}

logging.level:
        root: info
        ru.itone: debug

environment:
        # Название окружения
        name: ${ENVIRONMENT_NAME:test}
        # Папка для ошибочных файлов
        error-folder: ${ENVIRONMENT_ERROR_FOLDER:error}

zookeeper:
        # Адрес сервера zookeeper
        connection-string: ${ZK_CONNECTION:localhost}
        # Таймаут сессии
        session-timeout-ms: ${ZK_SESSION_TIMEOUT_MS:30000}
        # Таймаут подключения
        connection-timeout-ms: ${ZK_CONNECTION_TIMEOUT_MS:86400000}
        chroot: ${ZOOKEEPER_DS_CHROOT:/adapter}

migration:
        enabled: ${MIGRATION_ENABLE:false}

prostore-rest-client:
        host: ${PS_HOST:localhost}
        port: ${PS_PORT:9195}
        http:
                max-pool-size: ${PS_MAX_POOL_SIZE:8}

prostore:
        zookeeper:
                # Адрес сервера zookeeper для загрузки данных в простор
                connection-string: ${ZK_PROSTORE_CONNECTION:locahost:2181}

validation:
        enable: ${VALIDATION_ENABLE:true}
        rest-uploader-url: ${REST_UPLOADER_URL:http://localhost:8081}
        # обязательность использования ФЛК
        mandator: ${VALIDATION_MANDATOR:false}

upload:
        # требуется токен для аутентификации на rest-uploader
        jwt-auth: ${JWT_AUTH:false}

kafka:
        create-topic:
                # Количество партиций на загрузку через EDML
                num-partitions: ${EDML_UPLOAD_NUM_PARTITIONS:1}
                # Фактор репликации при создании топика
                replication-factor: ${EDML_UPLOAD_REPLICATION_FACTOR:1}
        topic:
                # Топик для журналирования
                journal-log: journal.log
                flk-log: flk.log
        consumer:
                # Количество партиций на выгрузку через EDML
                num-partitions: ${EDML_DOWNLOAD_NUM_PARTITIONS:1}
                property:
                        bootstrap.servers: *kafkaUrl
                        group.id: csv-uploader
                        auto.offset.reset: earliest
                        enable.auto.commit: true
        producer:
                property:
                        bootstrap.servers: *kafkaUrl

csv-parser:
        separator: ${CSV_PARSER_SEPARATOR:;}
        quote-char: ${CSV_PARSER_QUOTE_CHAR:"}
        escape-char: ${CSV_PARSER_ESCAPE_CHAR:'}
        field-as-null: ${CSV_PARSER_FIELD_AS_NULL:EMPTY_SEPARATORS}

metrics:
        port: ${METRICS_PORT:9837}

backup:
        zk-path: ${CSV_UPLOADER_BACKUP_ZK_PATH:/${environment.name}/csv-uploader/config}
        commandTopic: ${BACKUP_COMMAND_TOPIC:adapter.command}
        backupTopic: ${BACKUP_TOPIC:adapter.backup}
        statusTopic: ${STATUS_TOPIC:adapter.status}
        kafka:
                consumer:
                        property:
                                bootstrap.servers: ${KAFKA_BOOTSTRAP_SERVERS:localhost}
                                group.id: ${CSV_UPLOADER_BACKUP_GROUP_ID:csv_uploader_adapter_command}
                                auto.offset.reset: latest
                producer:
                        property:
                                bootstrap.servers: ${KAFKA_BOOTSTRAP_SERVERS:localhost}

jet-connector:
        use: ${JET_CONNECTOR_USE:false}

3. Параметры конфигурации

Настройка конфигурации CSV-uploader осуществляется путем редактирования параметров настроек в файле application.yml.

Некоторые настройки доступны для редактирования через пользовательский интерфейс модуля, например, Настройка отображения количества записей и Запуск по расписанию в Журнале операций .

В файле конфигурации CSV-uploader могут быть настроены следующие секции:

• kafkaUrl - URL для доступа к Kafka;

• http-server - настройки порта подключения;

• send - настройка отправки файлов;

• file-size - ограничение на размер отправляемого файла (мегабайты);

• logging.level - настройка сохранения лог-файла;

• environment - определяет значение среды разработки;

• zookeeper - настройка подключения Zookeeper;

• migration - настройка миграции зукипера для задачи бекапирования;

• prostore-rest-client - блок параметров конфигурирования взаимодействия с ProStore;

• prostore - адрес сервера zookeeper для загрузки данных в ProStore;

• validation - включение/выключение механизма валидации загрузки с помощью модуля REST-Uploader;

• upload - требование токена для аутентификации на REST-Uploader;

• kafka - настройка подключения к шине данных Apache Kafka;

• csv-parser - настройка парсинга CSV;

• metrics - настройка получения метрик;

• backup - настройка бэкапирования модуля;

• jet-connector - подготовлен для оптимизации задержек записи.

3.1. Секция kafkaUrl

В секция kafkaUrl указывается URL-адрес для доступа к Apache Kafka (ProStore).

Например:

.kafkaUrl: &kafkaUrl ${KAFKA_BOOTSTRAP_SERVERS:dev-dtm-one05.ru-central1.internal:9092}

Параметры конфигурации

KAFKA_BOOTSTRAP_SERVERS - URL-адрес для доступа к Apache Kafka (ProStore).

3.2. Секция http

Секция http предназначена для настройки порта и протокола передачи данных (одно из значений http или https).

Например:

http:
        port: ${HTTP_PORT:8080}
        enabled: ${HTTP_ENABLED:true}

Параметры конфигурации

• port - порт для старта веб-сервера, например HTTP_PORT:8080;

• enabled - статус включения/отключения веб-сервера, например HTTP_ENABLED:true.

3.3. Секция query-executor

В секции query-executor настраивается получение входящих запросов.

Например:

query-executor:
        reader-pool-size: ${READER_POOL_SIZE:20}
        rest-pool-size: ${REST_POOL_SIZE:5}
        writer-pool-size: ${WRITER_POOL_SIZE:1}
        kafka-pool-size: ${KAFKA_POOL_SIZE:20}

Параметры конфигурации

• reader-pool-size - размер пула на чтение, например: READER_POOL_SIZE:20;

• rest-pool-size - размер пула на REST, например: REST_POOL_SIZE:5;

• writer-pool-size - размер пула на запись, например: WRITER_POOL_SIZE:1;

• kafka-pool-size - размер kafka пула на отправку, например KAFKA_POOL_SIZE:20.

3.4. Секция send

В секции send настраивается отправка файлов.

Например:

send:
        chunk-row-count: ${CHUNK_ROW_COUNT:1000}
        file-buffer-size: ${FILE_BUFFER_SIZE:1048576}
        read-job-count: ${READ_JOB_COUNT:4}
        serialize-channel-size: ${SERIALIZE_CHANNEL_SIZE:20}
        serialize-job-count: ${SERIALIZE_JOB_COUNT:4}
        send-channel-size: ${SEND_CHANNEL_SIZE:20}
        send-job-count: ${SEND_JOB_COUNT:4}

Параметры конфигурации:

• chunk-row-count - размер отправляемой порции данных, например CHUNK_ROW_COUNT:100;

• file-buffer-size - размер буфера на чтение файла, например FILE_BUFFER_SIZE:1048576;

• read-job-count - количество Job на чтение, например READ_JOB_COUNT:4;

• serialize-channel-size - размер Channel для сериализации, например SERIALIZE_CHANNEL_SIZE:20;

• serialize-job-count - количество задач на сериализацию, например SERIALIZE_JOB_COUNT:4;

• send-channel-size - размер задач на отправку, например SEND_CHANNEL_SIZE:20;

• send-job-count - количество задач на отправку, например SEND_JOB_COUNT:4;

3.5. Секция file-size

Секция file-size отвечает за ограничение на размер отправляемого файла (указывется в мегабайтах).

file-size:
        #
        restriction: ${SEND_FILE_SIZE_RESTRICTION:1024}

Параметры конфигурации

• restriction - ограничение на размер отправляемого файла, например SEND_FILE_SIZE_RESTRICTION:1024.

3.6. Секция logging.level

В секции logging.level настраиваются записи логирования.

Например:

logging.level:
        root: info
        ru.itone: debug

3.7. Секция environment

В секции environment выбирается среда разработки (например, значение test, prod и т.д).

Например:

environment:
        name: ${ENVIRONMENT_NAME:test}
        error-folder: ${ENVIRONMENT_ERROR_FOLDER:error}

Параметры конфигурации

• name - название окружения, например ENVIRONMENT_NAME:test;

• error-folder - папка для ошибочных файлов, например ENVIRONMENT_ERROR_FOLDER:error.

3.8. Секция zookeeper

В секции zookeeper настраиваются параметры подключения к серверу Zookeeper.

Например:

zookeeper:
        connection-string: ${ZK_CONNECTION:t5-ads-02.ru-central1.internal}
        session-timeout-ms: ${ZK_SESSION_TIMEOUT_MS:30000}
        connection-timeout-ms: ${ZK_CONNECTION_TIMEOUT_MS:86400000}

Параметры конфигурации

• connection-string - адрес сервера Zookeeper, например ZK_CONNECTION:t5-ads-02.ru-central1.internal;

• session-timeout-ms - таймаут сессии, например ZK_SESSION_TIMEOUT_MS:30000;

• connection-timeout-ms - таймаут подключения, например ZK_CONNECTION_TIMEOUT_MS:86400000.

3.9. Секция migration

В секции migration реализована настройка миграции зукипера для задачи бекапирования.

Например:

migration:
        enabled: ${MIGRATION_ENABLE:false}

Параметры настроек

• enabled - включение миграции (по умолчанию выключена), например {MIGRATION_ENABLE:false}.

3.10. Секция prostore-rest-client

В секции prostore-rest-client реализован блок параметров конфигурирования взаимодействия с ProStore.

Например:

prostore-rest-client:
        host: ${PS_HOST:localhost}
        port: ${PS_PORT:9195}
        http:
                max-pool-size: ${PS_MAX_POOL_SIZE:8}

Параметры настроек

• host - адрес Prostore, например PS_HOST:localhost;

• port - порт Prostore, например PS_PORT:9195;

• max-pool-size - максимальное число подключений к Prostore, например PS_MAX_POOL_SIZE:8.

3.11. Секция prostore

В секции prostore указывается адрес сервера zookeeper для загрузки данных в Prostore.

Например:

prostore:
        zookeeper:
                connection-string: ${ZK_PROSTORE_CONNECTION:locahost:2181}

Параметры настроек

• connection-string - адрес сервера Zookeeper для загрузки данных в Prostore, например ZK_PROSTORE_CONNECTION:locahost:2181.

3.12. Секция validation

В секции validation реализован механизм настройки валидации ФЛК.

Например:

enable: ${VALIDATION_ENABLE:true}
rest-uploader-url: ${REST_UPLOADER_URL:http://localhost:8081}
mandator: ${VALIDATION_MANDATOR:false}

Параметры конфигурации

• enable - валидация включена (по умолчанию), например {VALIDATION_ENABLE:true};

• rest-uploader-url - URL к сервису rest-uploader для выполнения валидации, например {REST_UPLOADER_URL:http://localhost:8081};

• mandator - флаг использования ФЛК, например {VALIDATION_MANDATOR:false}.

3.13. Секция upload

В секции upload реализована настройка требования токена для аутентификации на REST-Uploader (если true, то при переключении на вкладку Загрузка появляется модальное окно для задания токена в текстовом виде и кнопка Сохранить).

Например:

upload:
        jwt-auth: ${JWT_AUTH:false}

Параметры конфигурации

• jwt-auth - требование токена для аутентификации на REST-Uploader, например {JWT_AUTH:false}.

3.14. Секция kafka

В секции kafka настраиваются параметры подключения к шине данных Apache Kafka.

Например:

kafka:
        create-topic:
                num-partitions: ${EDML_UPLOAD_NUM_PARTITIONS:1}
                replication-factor: ${EDML_UPLOAD_REPLICATION_FACTOR:1}
        topic:
                journal-log: journal.log
        consumer:
                num-partitions: ${EDML_DOWNLOAD_NUM_PARTITIONS:1}
                property:
                        bootstrap.servers: *kafkaUrl
                        group.id: csv-uploader
                        auto.offset.reset: earliest
                        enable.auto.commit: true
        producer:
                property:
                        bootstrap.servers: *kafkaUrl

Параметры конфигурации

• num-partitions - количество партиций на загрузку через EDML, например EDML_UPLOAD_NUM_PARTITIONS:1;

• replication-factor - фактор репликации при создании топика, например EDML_UPLOAD_REPLICATION_FACTOR:1.

3.15. Секция csv-parser

Внимание

При загрузке файлов с форматно-логическим контролем, важно, чтобы настройки секции csv-parser были одинаковыми в модулях CSV-Uploader(если используется его UI),REST-Uploader и DATA-Uploader.

Секция csv-parser - настройка парсинга CSV.

Например:

csv-parser:
        separator: ${CSV_PARSER_SEPARATOR:;}
        quote-char: ${CSV_PARSER_QUOTE_CHAR:"}
        escape-char: ${CSV_PARSER_ESCAPE_CHAR:'}
        field-as-null: ${CSV_PARSER_FIELD_AS_NULL:EMPTY_SEPARATORS}

Параметры конфигурации

• separator - символ разделителя полей, например CSV_PARSER_SEPARATOR:;;

• quote-char - символ кавычки, например CSV_PARSER_QUOTE_CHAR:";

• escape-char - символ экранирования, например CSV_PARSER_ESCAPE_CHAR:';

• field-as-null - способ определения null поля, например CSV_PARSER_FIELD_AS_NULL:EMPTY_SEPARATORS.

Дополнительное описание параметров

1. Параметр CSV_PARSER_ESCAPE_CHAR работает следующим образом: если символ экранирования и символ кавычки равны ", то будет использован RFC4180Parser, который считывает все символы между двумя двойными кавычками, при этом двойная кавычка в тексте поля должна быть экранирована двойной кавычкой (Например "поле, ""содержащее двойную кавычку""" будет считано как поле, "содержащее двойную кавычку"). В противном случае будет использован CSVParser, использующий символ экранирования для обозначения «непечатаемых символов».

2. Параметр CSV_PARSER_FIELD_AS_NULL может принимать следующие значения:

   • EMPTY_SEPARATORS - два разделителя полей (см. csv-parser/separator) подряд считаются null. Например: строка [aaa,,ccc] содержит значения [«aaa», null, «bbb»], а строка [aaa,»»,ccc] содержит значения [«aaa», «», «bbb»].

   • EMPTY_QUOTES - два «ограничителя строки» (см. csv-parser/escape-char) подряд считаются null. Например: строка [aaa,»»,ccc] содержит значения [«aaa», null, «bbb»], а строка [aaa,,ccc] содержит значения [«aaa», «», «bbb»].

   • BOTH - оба варианта (см. EMPTY_SEPARATORS и EMPTY_QUOTES) считаются null. Например: обе строки [aaa,»»,ccc] и [aaa,,bbb] содержат одинаковое значение [«aaa», null, «bbb»].

   • NEITHER - ни один из вариантов (см. EMPTY_SEPARATORS и EMPTY_QUOTES) не считается null. Например: обе строки [aaa,»»,ccc] и [aaa,,bbb] содержат одинаковое значение [«aaa», «», «bbb»].

3.16. Секция metrics

Секция metrics предназначена для настройки параметров метрик.

Например:

metrics:
  port: ${METRICS_PORT:9837}

Параметры конфигурации

• port - Порт для метрик, например METRICS_PORT:9837.

3.17. Секция backup

Секция backup предназначена для настроек бекапирования модуля.

Например:

backup:
  zk-path: ${COUNTER_BACKUP_ZK_PATH:/${environment.name}/counter-provider/counters}
  commandTopic: ${BACKUP_COMMAND_TOPIC:adapter.command}
  backupTopic: ${BACKUP_TOPIC:adapter.backup}
  statusTopic: ${STATUS_TOPIC:adapter.status}
  kafka:
    consumer:
      property:
        bootstrap.servers: ${KAFKA_BOOTSTRAP_SERVERS:localhost:9092}
        group.id: ${COUNTER_BACKUP_GROUP_ID:counter_provider_adapter_command}
        auto.offset.reset: latest
    producer:
      property:
        bootstrap.servers: ${KAFKA_BOOTSTRAP_SERVERS:localhost:9092}

Параметры настроек

• zk-path - путь к корневой ноде zookeeper для бэкапирования, например {COUNTER_BACKUP_ZK_PATH:/${environment.name}/counter-provider/counters};

• commandTopic - топик команд бэкапирования, например: {BACKUP_COMMAND_TOPIC:adapter.command};

• backupTopic - топик для отправки забэкапированных данных, например: {BACKUP_TOPIC:adapter.backup};

• statusTopic - топик для отправки статусов бэкапирования, например: {STATUS_TOPIC:adapter.status}.

3.18. Секция jet-connector

Секция jet-connector предназначена для оптимизации задержек записи данных.

Например:

jet-connector:
use: ${JET_CONNECTOR_USE:false}

Параметры настроек

• use - флаг активации jet-connector, например {JET_CONNECTOR_USE:false};

Таблица 3.3 Область применения

Режим/СУБД	ADB	ADP	ADQM	ADG
Чтение	Нет	Нет	Нет	Нет
Запись	В перспективе	Да	Нет	Нет

Чтобы воспользоваться Jet-коннектор требуется вместо upload external table создавать readable  external table, указывающую на топик, как отражено в документации Prostore

В модуль MPPW не передается информация о том, в какую БД физически будут загружаться данные, синтаксис Простора един для всех поддерживаемых СУБД. Конкретная СУБД указывается в настройках Простора.

Даже если загрузка данных выполняется в более чем одну базу, на работе адаптеров это не сказывается.

При формировании запросов с табличными параметрами, в том числе при регистрации РЗ, необходимо явно перечислять поля таблиц, по которым будет выполняться фильтрация записей или объединение таблиц.

Переключение с jet-connector на kafka postgres writer и наоборот допускается при завершенных операциях загрузки данных.

Предупреждение

Jet-коннектор в настоящее время применим для ADP, что делает его применимым, только для иснталляций одной единственной СУБД ADP. Это ограничение остается на уровне документации при использовании Jet-коннектор с другими базами должна появиться ошибка.