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:
        # Признак использования rest-api для взаимодействия с простором. Если false - будет использоваться jdbc драйвер
        enabled: ${PS_REST_CLIENT_ENABLED:true}
        host: ${PS_HOST:localhost}
        port: ${PS_PORT:9195}
        http:
                max-pool-size: ${PS_MAX_POOL_SIZE:8}

prostore:
        # subprotocol для jdbc
        subprotocol: ${PS_SUBPROTOCOL:prostore}
        # Драйвер для подключения к простору
        driver: ${PS_DRIVER:ru.datamart.prostore.jdbc.Driver}
        # Подключение к Prostore
        host: ${PS_HOST:localhost}
        # Prostore порт
        port: ${PS_PORT:9195}
        # Prostore начальный размер пула ру
        initialPoolSize: ${PS_INITIAL_POOL_SIZE:5}
        # Prostore максимальный размер пула
        maxPoolSize: ${PS_MAX_POOL_SIZE:8}
        additional-props:
                key: value
        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}
        kafka-consumer-group-name: ${JET_CONNECTOR_GROUP_NAME:csv-uploader}
        kafka-poll-duration-ms: ${JET_POLL_DURATION:200}
        kafka-poll-buffer-size: ${JET_POLL_BUFFER:2000}
        db-records-buffer-size: ${JET_DB_RECORDS_BUFFER:2000}

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

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

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

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

kafkaUrl - URL для доступа к Kafka;
http-server - настройки порта подключения;
send - настройка отправки файлов;
logging.level - настройка сохранения лог-файла;
environment - определяет значение среды разработки;
zookeeper - настройка подключения Zookeeper;
migration - настройка миграции зукипера для задачи бекапирования;
prostore-rest-client - блок параметров конфигурирования взаимодействия с ProStore. Если false - будет использоваться JDBC-драйвер;
prostore - настройка подключения к 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 - порт для старта веб-сервера;
enabled - статус включения/отключения веб-сервера.

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}
        file-size-restriction: ${SEND_FILE_SIZE_RESTRICTION:1024}

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

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;
file-size-restriction - ограничение на размер оправляемого файла (мегабайты), например SEND_FILE_SIZE_RESTRICTION:1024.

3.5. Секция logging.level

Секция logging.level определяет настройки записи логирования.

Например:

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

3.6. Секция 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.7. Секция 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.8. Секция migration

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

Например:

migration:
        enabled: ${MIGRATION_ENABLE:false}

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

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

3.9. Секция prostore-rest-client

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

Например:

prostore-rest-client:
        enabled: ${PS_REST_CLIENT_ENABLED:true}
        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.10. Секция prostore

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

Например:

prostore:
        subprotocol: ${PS_SUBPROTOCOL:prostore}
        driver: ${PS_DRIVER:ru.datamart.prostore.jdbc.Driver}
        host: ${PS_HOST:dev-dtm-one05.ru-central1.internal}
        port: ${PS_PORT:9090}
        initialPoolSize: ${PS_INITIAL_POOL_SIZE:5}
        maxPoolSize: ${PS_MAX_POOL_SIZE:8}
        additional-props:
        key: value
        zookeeper:
                connection-string: ${ZK_PROSTORE_CONNECTION:dev-dtm-one05.ru-central1.internal:2181}

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

subprotocol - subprotocol для jdbc, например PS_SUBPROTOCOL:prostore;
driver - драйвер для подключения к Prostore, например PS_DRIVER:ru.datamart.prostore.jdbc.Driver;
host - подключение к Prostore, например PS_HOST:dev-dtm-one05.ru-central1.internal;
port - Prostore порт, например PS_PORT:9090;
initialPoolSize - начальное число подключений к Prostore, например PS_INITIAL_POOL_SIZE:5;
maxPoolSize - максимальное число подключений к Prostore, например, например PS_MAX_POOL_SIZE:8;
connection-string - адрес сервера zookeeper для загрузки данных в Prostore, например ZK_PROSTORE_CONNECTION:dev-dtm-one05.ru-central1.internal:2181.

3.11. Секция 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.12. Секция upload

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

Например:

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

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

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

3.13. Секция 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.14. Секция 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.

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

Параметр CSV_PARSER_ESCAPE_CHAR работает следующим образом: если символ экранирования и символ кавычки равны ", то будет использован RFC4180Parser, который считывает все символы между двумя двойными кавычками, при этом двойная кавычка в тексте поля должна быть экранирована двойной кавычкой (Например "поле, ""содержащее двойную кавычку""" будет считано как поле, "содержащее двойную кавычку"). В противном случае будет использован CSVParser, использующий символ экранирования для обозначения «непечатаемых символов».
Параметр 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.15. Секция metrics

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

Например:

metrics:
  port: ${METRICS_PORT:9837}

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

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

3.16. Секция 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.17. Секция jet-connector

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

Например:

jet-connector:
use: ${JET_CONNECTOR_USE:false}
kafka-consumer-group-name: ${JET_CONNECTOR_GROUP_NAME:csv-uploader}
kafka-poll-duration-ms: ${JET_POLL_DURATION:200}
kafka-poll-buffer-size: ${JET_POLL_BUFFER:2000}
db-records-buffer-size: ${JET_DB_RECORDS_BUFFER:2000}

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

use - флаг активации jet connector’а, например {JET_CONNECTOR_USE:false};
kafka-consumer-group-name - имя консьюмер-группы, загружающей данные, например: {JET_CONNECTOR_GROUP_NAME:csv-uploader};
kafka-poll-duration-ms - максимальное время одного прохода вычитки из топика в мс, например: {JET_POLL_DURATION:200};
kafka-poll-buffer-size - максимальный размер буфера прочитанных, но еще не загруженных сообщений, например: {JET_POLL_BUFFER:2000};
db-records-buffer-size - максимальное количество строк, загружаемых в таблицу за один проход, например: {JET_DB_RECORDS_BUFFER:2000}.