Vector является очень гибким агрегатором сообщений и поддерживает собственный язык для обработки событий - VRL.

Поэтому предлагаю без лишних слов перейти к изучению его возможностей. Давайте напишем простейшую конфигурацию с использованием VRL, которая будет удалять поле из JSON файла

На входе будут такие данные:

{
  "field1": "foo",
  "field2": "bar"
}

Наша первая конфигурация на VRL будет выглядеть так:

sources:
  file_input:
    type: file
    include:
      - /opt/habr/vector/intro.json
    ignore_checkpoints: true

transforms:
  parse_file:
    type: remap
    inputs:
      - file_input
    source: |

        .message = parse_json!(.message)
        .field1 = .message.field1

        del(.host)
        del(.file)
        del(.message)
        del(.timestamp)
        del(.source_type)

sinks:
  test_output:
    type: file
    inputs:
      - parse_file
    path: /opt/habr/vector/intro_out.json
    encoding:
      codec: json

И на выходе мы получим такую данные:

{
  "field1": "foo"
}

Зачем в нашем трансформе такая часть с удалением других полей?

        del(.host)
        del(.file)
        del(.message)
        del(.timestamp)
        del(.source_type)

Дело в том, что на источник данных в нашем примере - file, который содержит помимо самой строки, ещё и другие метаданные в событии.
Вот как бы выглядел наш вывод, если оставить эти поля:

{
  "field1": "foo",
  "file": "/opt/habr/vector/intro.json",
  "host": "test-mon",
  "message": {
    "field1": "foo",
    "field2": "bar"
  },
  "source_type": "file",
  "timestamp": "2025-03-17T13:09:20.653470883Z"
}

Здесь стоит обратить внимание, что наше поле field1 встречается дважды. Правда в первом случае - это отдельный объект, поскольку мы объявили его внутри нашего трансформа, а во втором случае - это часть message, то есть содержимое нашей строки.

Полезной практикой при написании VRL будет создание отдельного файла под VRL. В нашем примере VRL получился совсем короткий, но бывают случаи, когда один VRL растягивается на 100, 200 и 300 строк. Поэтому хранить всё в одном файле будет неудобно, есть риск допустить ошибки в конфигурации, а также теряется читабельность.

Чтобы вынести VRL в отдельный файл, достаточно указать параметр "file" в transform

sources:
  file_input:
    type: file
    include:
      - /opt/habr/vector/intro.json
    ignore_checkpoints: true

transforms:
  parse_file:
    type: remap
    inputs:
      - file_input
    file: /etc/vector/habr/intro.vrl

sinks:
  test_output:
    type: file
    inputs:
      - parse_file
    path: /opt/habr/vector/intro_out.json
    encoding:
      codec: json

Теперь наша конфигурация выглядит компактнее, а с самим VRL можно работать в отдельном файле.

2. Синтаксис VRL

Теперь изучив базовую работу с VRL, можно изучить его синтаксис, чтобы начать работать с Vector более гибко.

Переменные

Переменные в VRL могут содержать числовые, строковые и логические значения. Их можно присвоить напрямую:

$ a=5                                                                
5

$ b = "some text"
"some text"

$ pi = 3.14
3.14

$ k = true
true

Либо же можно присвоить значение как результат выполнения функции:

$ c = snakecase(b)
"some_text"

Также с числовыми переменными можно выполнять базовые математические операции:

$ a = 5 + 5
10

$ a = 5 - 2
3

$ a = 5 * 5
25

$ a = 5 / 5
1

$ a = 7 / 5
1.4

$ a = mod(5, 3)
2

А также мы можем объединять вместе строковые переменные:

$ str1 = "some text"
"some text"

$ str2 = "other text"
"other text"

$ str1 + " and " + str2
"some text and other text"

Также стоит отметить, что переменные, начинающиеся с точки - это JSON поля, которые пойдут дальше в sink, а без неё - это обычные переменные, которые существуют в рамках трансформа.

Массивы

VRL поддерживает работу с массивами. Массивы могут хранить в себе любые типы данных, такие как переменные, числа, строки или JSON объекты:

$ a = 5
5

$ arr = ["text", 5, 3.14, true, a, {"foo": "bar"}]
["text", 5, 3.14, true, 5, { "foo": "bar" }]

Операторы и циклы

Полноценно VRL поддерживает только оператор if, как таковых других циклов по типу for или while здесь нет, привычных нам по большинству языков программирования, но в случае с for есть функция, которая выполняет схожий механизм, о котором поговорим чуть позже. А вот при попытке использовать цикл while, мы увидим лишь, что это ключевое слово зарезервировано для будущих нужд:

$ while

error[E205]: reserved keyword
  ┌─ :1:1
  │
1 │ while
  │ ^^^^^
  │ │
  │ this identifier name is reserved for future use in the language
  │ use a different name instead
  │
  = see language documentation at https://vrl.dev
  = try your code in the VRL REPL, learn more at https://vrl.dev/examples

В целом, полезно ознакомиться со списком зарезервированных слов, чтобы избежать конфликтов - https://vector.dev/docs/reference/vrl/expressions/#keywords

Итак, начнём с простого оператора проверки условия if - else.
Здесь мы имеем вполне знакомый синтаксис:

if *условие* {
	*дейстиве*
} else if *условие* {
	*дейтсвие*
} else {
	*действие*
}

И попробуем это выполнить на реальном примере:

$ a = 3
if (a == 10) {
  "first block"
} else if (a <= 10) {
  "second block"
} else {
  "third block"
}
"second block"

Теперь перейдём к перебору массивов. Как я уже писал, на данный момент VRL не поддерживает цикл for и есть лишь функция, которая выполняет схожий механизм.

Рассмотрим функцию for_each. Для примера возьмём такой массив:

	.arr = ["my", "first", "array", "loop"]

И теперь попробуем применить функцию upcase для каждого элемента массива с помощью for_each:

$ .new_arr = []
$
$ for_each(.arr) -> |_index, value|{
    .new_arr = push(.new_arr, upcase(value))
}
$
$ .new_arr
["MY", "FIRST", "ARRAY", "LOOP"]
$ .arr = del(.new_arr)
["MY", "FIRST", "ARRAY", "LOOP"]

Как видим, здесь всё не так просто. Для начала нам надо создать новый массив, который будет хранить обработанные значения. Затем мы пробегаемся по массиву .arr, используя for_each. Выражение |_index, value|, означает, что мы будет работать только со значениями массива, и индексы нас не интересуют, поэтому будет обработан каждый элемент массива. Далее мы добавляем новое обработанное значение в массив .new_arr с помощью функции push. После чего мы проверяем содержимое временного массива, присваиваем его нашему исходному массиву и удаляем временный.

У вас мог возникнуть вопрос, почему мы не можем просто использовать исходный массив, и применить функцию upscale к каждому элементу? Но, к сожалению, VRL не позволяет нам работать с массивом таким образом:

$ for_each(.arr) -> |index, value| {
    .arr[index] = upcase(value)
}

error[E203]: syntax error
  ┌─ :2:10
  │
2 │     .arr[index] = upcase(value)
  │          ^^^^^
  │          │
  │          unexpected syntax token: "Identifier"
  │          expected one of: "integer literal"
  │
  = see language documentation at https://vrl.dev
  = try your code in the VRL REPL, learn more at https://vrl.dev/examples

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

$ map_values(.arr) -> |value| { upcase!(value) }
["MY", "FIRST", "ARRAY", "LOOP"]

Теперь давайте рассмотрим аналогичное действие, но на это раз обработаем только чётные элементы массива .arr:

$ .new_arr = []
$
$ for_each(.arr) -> |index, value| {
    if mod(index, 2) == 0 {
        .new_arr = push(.new_arr, upcase(value))
    } else{
        .new_arr = push(.new_arr, value)
    }
}
$
$ .new_arr
["MY", "first", "ARRAY", "loop"]
$ .arr = del(.new_arr)
["MY", "first", "ARRAY", "loop"]

Здесь мы вновь создаём наш временный массив, который будет хранить обработанные значения. На это раз мы указываем index без нижнего подчёркивания и в теле for_each проверяем значение индекса. Если его остаток от деления на 2 равен 0, тогда этот элемент массива .arr чётный и мы применяем к нему функцию upscale. В противном случаем элемент массива .arr будет добавлен массиву .new_arr без изменений.

Обратите внимание, на блок else. Его обязательно надо добавить в наш цикл, поскольку без него нечётные элементы просто будут отброшены:

$ .new_arr
["MY", ARRAY"]

Обработка ошибок

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

Представим ситуацию, у нас есть файл с таким содержимым:

1,,3

И мы хотим разбить каждое число в отдельное поле и умножить его на 5, разделив их запятой. Для этого напишем такой VRL:

arr = split(.message, ",")

.var1 = to_int(arr[0]) * 5
.var2 = to_int(arr[1]) * 5
.var3 = to_int(arr[2]) * 5

А сам файл конфигурации будет выглядеть так:

sources:
  file_input:
    type: file
    include:
      - /opt/habr/vector/error.txt
    ignore_checkpoints: true

transforms:
  parse_file:
    type: remap
    inputs:
      - file_input
    file: /etc/vector/habr/errors.vrl

sinks:
  test_output:
    type: file
    inputs:
      - parse_file
    path: /opt/habr/vector/error.json
    encoding:
      codec: json

Теперь попробуем запустить vector с такой конфигурацией:

root@test-mon:/etc/vector/habr# vector --config errors.yaml 
2025-07-16T14:30:36.222710Z  INFO vector::app: Log level is enabled. level="info"
2025-07-16T14:30:36.223878Z  INFO vector::app: Loading configs. paths=["errors.yaml"]
2025-07-16T14:30:36.226896Z ERROR vector::topology::builder: Configuration error. error=Transform "parse_file": 
error[E103]: unhandled fallible assignment
  ┌─ :1:7
  │
1 │ arr = split(.message, ",")
  │ ----- ^^^^^^^^^^^^^^^^^^^^
  │ │     │
  │ │     this expression is fallible because at least one argument's type cannot be verified to be valid
  │ │     update the expression to be infallible by adding a `!`: `split!(.message, ",")`
  │ │     `.message` argument type is `any` and this function expected a parameter `value` of type `string`

Во-первых, мы видим, что vector сразу ругается на функцию split, поскольку такая запись может привести к ошибкам. В нашем случае мы полностью уверены в нашем выборе, поэтому обозначим функцию split с символом «!», чтобы вектор проигнорировал потенциально опасную запись и продолжил работу:

arr = split!(.message, ",")

И попробуем запустить vector:

root@test-mon:/etc/vector/habr# vector --config errors.yaml 
2025-07-16T14:35:09.037986Z  INFO vector::app: Log level is enabled. level="info"
2025-07-16T14:35:09.039181Z  INFO vector::app: Loading configs. paths=["errors.yaml"]
2025-07-16T14:35:09.042971Z ERROR vector::topology::builder: Configuration error. error=Transform "parse_file": 
error[E103]: unhandled fallible assignment
  ┌─ :3:9
  │
3 │ .var1 = to_int(arr[0]) * 5
  │ ------- ^^^^^^^^^^^^^^^^^^ this expression is fallible because at least one argument's type cannot be verified to be valid

И вновь непорядок, вновь vector падает из-за ошибки. На это раз ему не нравится функция to_int.
Представим, что мы очень уверенные в себе люди, и так же добавим «!» к функции

.var1 = to_int!(arr[0]) * 5
.var2 = to_int!(arr[1]) * 5
.var3 = to_int!(arr[2]) * 5

Наверное теперь всё должно пройти хорошо и vector не свалится в ошибку:

root@test-mon:/etc/vector/habr# vector --config errors.yaml 
2025-07-16T14:36:22.005848Z  INFO vector::app: Log level is enabled. level="info"
2025-07-16T14:36:22.006867Z  INFO vector::app: Loading configs. paths=["errors.yaml"]
2025-07-16T14:36:22.010056Z  INFO vector::topology::running: Running healthchecks.
2025-07-16T14:36:22.010141Z  INFO vector::topology::builder: Healthcheck passed.
2025-07-16T14:36:22.010142Z  INFO vector: Vector has started. debug="false" version="0.45.0" arch="x86_64" revision="063cabb 2025-02-24 14:52:02.810034614"
2025-07-16T14:36:22.010167Z  INFO vector::app: API is disabled, enable by setting `api.enabled` to `true` and use commands like `vector top`.
2025-07-16T14:36:22.010197Z  INFO source{component_kind="source" component_id=file_input component_type=file}: vector::sources::file: Starting file server. include=["/opt/habr/vector/error.txt"] exclude=[]
2025-07-16T14:36:22.010491Z  INFO source{component_kind="source" component_id=file_input component_type=file}:file_server: file_source::checkpointer: Loaded checkpoint data.
2025-07-16T14:36:22.010623Z  INFO source{component_kind="source" component_id=file_input component_type=file}:file_server: vector::internal_events::file::source: Found new file to watch. file=/opt/habr/vector/error.txt
2025-07-16T14:36:22.010881Z ERROR transform{component_kind="transform" component_id=parse_file component_type=remap}: vector::internal_events::remap: Mapping failed with event. error="function call error for \"to_int\" at (65:80): Invalid integer \"\": cannot parse integer from empty string" error_type="conversion_failed" stage="processing" internal_log_rate_limit=true

Вроде бы всё хорошо, в этот раз vector не упал. Но в логе можно увидеть одно сообщение:

2025-07-16T14:36:22.010881Z ERROR transform{component_kind="transform" component_id=parse_file component_type=remap}: vector::internal_events::remap: Mapping failed with event. error="function call error for \"to_int\" at (65:80): Invalid integer \"\": cannot parse integer from empty string" error_type="conversion_failed" stage="processing" internal_log_rate_limit=true

В этой записи, мы видим, что на вход функции to_int поступило пустое число. Что же тогда vector записал в файл вывода?

{"file":"/opt/habr/vector/error.txt","host":"test-mon","message":"1,,3","source_type":"file","timestamp":"2025-07-16T14:36:22.010766422Z"}

А в выходном файле у нас записано событие целиком, поскольку оно обработалось с ошибкой.

Этот пример отчётливо показывает, что не стоит злоупотреблять знаками «!», а нужно их использовать только в том случае, если вы уверены, что ваша форма записи отработает корректно.

Но как же нам тогда поступить в этой ситуации? Для начала стоит отметить, зачем мы используем функцию to_int. Дело в том, что после обработки функцией split, мы получаем на выходе все элементы массива arr в string. Поэтому мы не можем просто умножить на 5, нам нужно привести данные к int. И в нашем исходном файле есть пропущенное число, поэтому такой случай надо отдельно обозначить:

.var1, err = to_int(arr[0]) * 5
.var2, err = to_int(arr[1]) * 5
.var3, err = to_int(arr[2]) * 5

Посмотрим, как теперь выглядят строки в выходном файле:

{"file":"/opt/habr/vector/error.txt","host":"test-mon","message":"1,,3","source_type":"file","timestamp":"2025-07-16T17:29:05.314777030Z","var1":5,"var2":0,"var3":15}

Видно, что теперь у нас обработались наши поля, а поле с пустым значением приравнялось у нулю. Всё потому что arr[1] у нас null и при преобразовании через `to_int, мы получаем 0

Применяем VRL на практике

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

Возьмём за основу такие данные с такой структурой полей:

"account_receiver": "234567891012",
"account_sender": "750",
"auth_channel_end": "PS",
"auth_channel_start": "PS",
"auth_context_end": "PS,43604000B000DA02,2025-02-14T10:10:28+03:00",
"auth_context_start": "PS,43604000B000DA02,2025-02-14T10:10:28+03:00",
"authorization_time": "2025-01-21 08:10:35",
"branch_code": "123456789101",
"channel_type": "3",
"compliance_rules": "345678912101,91,[],[],,AC",
"counterparty_id_a": "",
"counterparty_id_b": "urn:86661307-056805-0",
"customer_external_id": "436046005035947",
"customer_id": "234567891012",
"customer_internal_id": "",
"device_type": "1",
"entry_time_end": "2025-02-14 10:10:28",
"entry_time_start": "2025-02-14 10:10:28",
"entrypoint_node_end": "43604000B000DA02",
"entrypoint_node_start": "43604000B000DA02",
"forwarding": "",
"forwarding_reason": "",
"operation_code": "234567891012",
"processing_code": "900",
"processing_time_sec": "55",
"security_flag": "",
"session_token_a": "jrccpjjjqhkcj2ar2k0srqrqa20qanc2@1.2.3.4",
"session_token_b": "678F728F12B9D00030BA8@2.3.4.5:1234",
"session_token_c": "",
"status_http": "",
"transaction_end_time": "2025-02-14 10:11:17",
"transaction_id": "51ab65e0-0805-498f-8508-3cc4cd459ffd",
"transaction_start_time": "2025-01-21 08:10:22",
"transaction_status_code": "16",
"transaction_type": "522",
"used_services_raw": "OIP,override,0,,,2025-02-14T10:10:23+03:00,,,,[];CAT,,0,,,2025-02-14T10:10:31+03:00,,,,[]",
"user_interaction_time_sec": "43"

И сами строки в сыром виде, будут выглядеть так:

2025-02-14T10:11:17+03:00;522;123456789101;51ab65e0-0805-498f-8508-3cc4cd459ffd;234567891012;2025-01-21T10:10:22+05:00;2025-01-21T10:10:35+05:00;234567891012;750;234567

891012;;;3;1;jrccpjjjqhkcj2ar2k0srqrqa20qanc2@1.2.3.4;678F728F12B9D00030BA8@2.3.4.5:1234;;;436046005035947;;urn:86661307-056805-0;PS,43604000B000DA02,2025-02-14T10:10:2

8+03:00;PS,43604000B000DA02,2025-02-14T10:10:28+03:00;"";900;"345678912101,91,[],[],,AC";"OIP,override,0,,,2025-02-14T10:10:23+03:00,,,,[];CAT,,0,,,2025-02-14T10:10:31+

03:00,,,,[]";16;;55;43;

Теперь представим себе такую задачу: наша банковская система генерирует данные в таком формате, а партнёру надо преобразовать данные в удобный для него формат. Нам необходимо будет поменять формат меток времени из RFC 3339 в простой DateTime формат, а также нам надо заменить разделители в compliance_rules и used_services_raw с «;» на «&»

Для этого набросаем такой VRL. Для начала нам надо разбить строку на части, поскольку количество элементов не фиксировано, и просто разбить через «;» не получится, то мы поделим сообщение на части с помощью регулярного выражения по ключевым моментам в каждом сообщении:

.message = string!(.message)
.message = parse_regex!(.message, r'^(?P<log_part1>.*?);"(?P<log_security_flag>)";(?P<log_processing_code>[^;]*);"(?P<log_compliance_rules>[^"]*)";"(?P<log_used_service
s_raw>[^"]*)";(?P<log_part2>.*)$')
.part1 = .message.log_part1
.security_flag = .message.log_security_flag
.processing_code = .message.log_processing_code
.compliance_rules = .message.log_compliance_rules
.used_services_raw = .message.log_used_services_raw
.part2 = .message.log_part2

Здесь мы разбиваем сообщение на несколько частей, чтобы в дальнейшем их обработать отдельно. Также мы сразу создаём уже вычлененные поля в отдельные поля.

Теперь перейдём к обработке первой части сообщения:

.part1 = string(.part1)
arr = split(.part1, ";")
.transaction_end_time, err = if arr[0] != "" {
    unix_timestamp = to_unix_timestamp(parse_timestamp!(arr[0], "%+"))
    unix_timestamp = to_int(unix_timestamp) + 10800
    format_timestamp(from_unix_timestamp!(unix_timestamp), "%Y-%m-%d %H:%M:%S")
} else {
    ""
}
.transaction_type = to_string(arr[1])
.branch_code = to_string(arr[2])
.transaction_id = to_string(arr[3])
.customer_id = to_string(arr[4])
.transaction_start_time, err = if arr[5] != "" {
    unix_timestamp = to_unix_timestamp(parse_timestamp!(arr[5], "%+"))
    unix_timestamp = to_int(unix_timestamp) + 10800
    format_timestamp(from_unix_timestamp!(unix_timestamp), "%Y-%m-%d %H:%M:%S")
} else {
    ""
}
.authorization_time, err = if arr[6] != "" {
    unix_timestamp = to_unix_timestamp(parse_timestamp!(arr[6], "%+"))
    unix_timestamp = to_int(unix_timestamp) + 10800
    format_timestamp(from_unix_timestamp!(unix_timestamp), "%Y-%m-%d %H:%M:%S")
} else {
    ""
}
.operation_code = to_string(arr[7])
.account_sender = to_string(arr[8])
.account_receiver = to_string(arr[9])
.forwarding = to_string(arr[10])
.forwarding_reason = to_string(arr[11])
.channel_type = to_string(arr[12])
.device_type = to_string(arr[13])
.session_token_a = to_string(arr[14])
.session_token_b = to_string(arr[15])
.session_token_c = to_string(arr[16])
.customer_internal_id = to_string(arr[17])
.customer_external_id = to_string(arr[18])
.counterparty_id_a = to_string(arr[19])
.counterparty_id_b = to_string(arr[20])
.auth_context_start = to_string(arr[21])
.auth_context_end = to_string(arr[22])

Разберём это детальнее. Для начала мы разбиваем первую часть с помощью «;» и записываем это в массив arr:

.part1 = string(.part1)
arr = split(.part1, ";")

После чего мы получаем мы преобразуем наш первый timestamp transaction_end_time в нужный формат.

.transaction_end_time, err = if arr[0] != "" {
    unix_timestamp = to_unix_timestamp(parse_timestamp!(arr[0], "%+"))
    unix_timestamp = to_int(unix_timestamp) + 10800
    format_timestamp(from_unix_timestamp!(unix_timestamp), "%Y-%m-%d %H:%M:%S")
} else {
    ""
}

Для начала мы проверяем, что первый элемент массива arr не пустой, в противном случае он останется пустым, после чего мы преобразуем нашу метку времени в обычный unix timestamp, преобразуем его в int и прибавляем к нему 10800 (3 часа). Это необходимо для того, чтобы не сбить наше время от таймзоны, поскольку на выходе у нас получится unix timestamp в GMT +0. И после этого мы преобразуем наш unix timestamp в нужный формат времени. Далее мы будем применять аналогичный способ для всех меток времени.
А также создаём отдельные поля:

.transaction_type = to_string(arr[1])
...
.auth_context_end = to_string(arr[22])

Аналогичным образом мы поступим и для auth_context_start и auth_context_end

.auth_context_start = string(.auth_context_start)
arr1 = split(.auth_context_start, ",")

.auth_channel_start = to_string(arr1[0])
.entrypoint_node_start = to_string(arr1[1])
.entry_time_start, err = if arr1[2] != "" {
    unix_timestamp = to_unix_timestamp(parse_timestamp!(arr1[2], "%+"))
    unix_timestamp = to_int(unix_timestamp) + 10800
    format_timestamp(from_unix_timestamp!(unix_timestamp), "%Y-%m-%d %H:%M:%S")
} else {
    ""
}


.auth_context_end = string(.auth_context_end)
arr2 = split(.auth_context_end, ",")

.auth_channel_end = to_string(arr2[0])
.entrypoint_node_end = to_string(arr2[1])
.entry_time_end, err = if arr2[2] != "" {
    unix_timestamp = to_unix_timestamp(parse_timestamp!(arr2[2], "%+"))
    unix_timestamp = to_int(unix_timestamp) + 10800
    format_timestamp(from_unix_timestamp!(unix_timestamp), "%Y-%m-%d %H:%M:%S")
} else {
     ""
}

Дальше мы поступим очень интересно. Внутри compliance_rules может идти список из нескольких правил, но в наем случае всегда одно. Мы будем менять разделитель «;» между правилами на «&»:

.compliance_rules = replace(.compliance_rules, ";", "&")

А также нам надо заменить запятые внутри квадратный скобок. Для этого мы воспользуемся replace_with, чтобы найти внутри compliance_rules блок с фигурными скобками и заменить внутри запятые:

.compliance_rules, err = replace_with(.compliance_rules, r'\[(.*?)\]') -> |m| {
    "[" + replace(m.captures[0], ",", "&") + "]"
}

Здесь передаём функции compliance_rules и задаём регулярное выражение для поиска блока с квадратными скобками. Полученное совпадение записывается в переменную m и уже внутри неё меняются запятые на «&».

Далее нам надо преобразовать метку внутри used_services_raw:

.used_services_raw = string(.used_services_raw)
.services_info = split(.used_services_raw, ";")
.parsed_services_info = []

for_each(.services_info) -> |_index, value| {
    next_service = split(value, ",")

    service_call_time, err = if next_service[5] != "" {
        unix_timestamp = to_unix_timestamp(parse_timestamp!(next_service[5], "%+"))
        unix_timestamp = to_int(unix_timestamp) + 10800
        format_timestamp(from_unix_timestamp!(unix_timestamp), "%Y-%m-%d %H:%M:%S")
    } else {
        ""
    }

    obj = {
        "service_name": to_string(next_service[0]),
        "service_mode": to_string(next_service[1]),
        "number_of_diversions": to_string(next_service[2]),
        "assoc_party_id": to_string(next_service[3]),
        "service_id": to_string(next_service[4]),
        "service_call_time": service_call_time,
        "number_of_participants": to_string(next_service[6]),
        "action_type": to_string(next_service[7]),
        "customer_group_tag": to_string(next_service[8]),
        "service_cost_info": to_string(next_service[9])
    }

    .parsed_services_info = push(.parsed_services_info, obj)
}

Внутри used_services_raw у нас может идти сколько угодно сервисов, поэтому мы разбиваем его с помощью «;» и записываем в массив service_info.
Далее мы парсим каждый сервис из массива, пробежавшись по нему функцией for_each. Внутри мы по знакомой схеме меняем timestamp, записываем это в словарь obj и записываем его в отдельное поле.

Затем мы идёт простая часть с вычленением полей из второй части сообщения:

.part2 = string(.part2)
arr3 = split(.part2, ";")
.transaction_status_code = to_string(arr3[0])
.status_http = to_string(arr3[1])
.processing_time_sec = to_string(arr3[2])
.user_interaction_time_sec = to_string(arr3[3])

И теперь нам остаётся собрать воедино все разбитые поля в .message, чтобы записать обработанную строку в выходной файл.

.message =      .transaction_end_time + ";" +
        .transaction_type + ";" +
        .branch_code + ";" +
        .transaction_id + ";" +
        .customer_id + ";" +
        .transaction_start_time + ";" +
        .authorization_time + ";" +
        .operation_code + ";" +
        .account_sender + ";" +
        .account_receiver + ";" +
        .forwarding + ";" +
        .forwarding_reason + ";" +
        .channel_type + ";" +
        .device_type + ";" +
        .session_token_a + ";" +
        .session_token_b + ";" +
        .session_token_c + ";" +
        .customer_internal_id + ";" +
        .customer_external_id + ";" +
        .counterparty_id_a + ";" +
        .counterparty_id_b + ";" +
        .auth_channel_start + "," +
        .entrypoint_node_start + "," +
        .entry_time_start + ";" +
        .auth_channel_end + "," +
        .entrypoint_node_end + "," +
        .entry_time_end + ";\"" + 
        .security_flag + "\";" +
        .processing_code + ";\"" +
        .compliance_rules + "\";\""

for_each(.parsed_services_info) -> |index, obj| {
    if index != 0 {
        .message = .message + "&"
    }
    .message = .message + 
               obj.service_name + "," +
               obj.service_mode + "," +
               obj.number_of_diversions + "," +
               obj.assoc_party_id + "," +
               obj.service_id + "," +
               obj.service_call_time + "," +
               obj.number_of_participants + "," +
               obj.action_type + "," +
               obj.customer_group_tag + "," +
               obj.service_cost_info
        }


.message = .message + "\";" +
                .transaction_status_code + ";" +
                .status_http + ";" +
                .processing_time_sec + ";" +
                .user_interaction_time_sec + ";"

Здесь мы поочерёдно прописываем в нужном порядке наши поля, добавляя к ним разделители.
Отдельно можно выделить секцию с parsed_services_info. Нам также надо заменить разделитель между сервисами на «&». Поэтому мы поочерёдно записываем содержимое словарей в message, а также добавляем перед ним «&». Но здесь, стоит заметить, что у нас добавлена проверка для первого элемента, чтобы перед ним не ставился разделитель, поскольку перед ним нет другого сервиса, иначе первый сервис тоже бы начинался с разделителя, что неверно.

Конечный VRL будет выглядеть так:

.message = string!(.message)
.message = parse_regex!(.message, r'^(?P<log_part1>.*?);"(?P<log_security_flag>)";(?P<log_processing_code>[^;]*);"(?P<log_compliance_rules>[^"]*)";"(?P<log_used_services_raw>[^"]*)";(?P<log_part2>.*)$')
.part1 = .message.log_part1
.security_flag = .message.log_security_flag
.processing_code = .message.log_processing_code
.compliance_rules = .message.log_compliance_rules
.used_services_raw = .message.log_used_services_raw
.part2 = .message.log_part2

.part1 = string(.part1)
arr = split(.part1, ";")
.transaction_end_time, err = if arr[0] != "" {
    unix_timestamp = to_unix_timestamp(parse_timestamp!(arr[0], "%+"))
    unix_timestamp = to_int(unix_timestamp) + 10800
    format_timestamp(from_unix_timestamp!(unix_timestamp), "%Y-%m-%d %H:%M:%S")
} else {
    ""
}
.transaction_type = to_string(arr[1])
.branch_code = to_string(arr[2])
.transaction_id = to_string(arr[3])
.customer_id = to_string(arr[4])
.transaction_start_time, err = if arr[5] != "" {
    unix_timestamp = to_unix_timestamp(parse_timestamp!(arr[5], "%+"))
    unix_timestamp = to_int(unix_timestamp) + 10800
    format_timestamp(from_unix_timestamp!(unix_timestamp), "%Y-%m-%d %H:%M:%S")
} else {
    ""
}
.authorization_time, err = if arr[6] != "" {
    unix_timestamp = to_unix_timestamp(parse_timestamp!(arr[6], "%+"))
    unix_timestamp = to_int(unix_timestamp) + 10800
    format_timestamp(from_unix_timestamp!(unix_timestamp), "%Y-%m-%d %H:%M:%S")
} else {
    ""
}
.operation_code = to_string(arr[7])
.account_sender = to_string(arr[8])
.account_receiver = to_string(arr[9])
.forwarding = to_string(arr[10])
.forwarding_reason = to_string(arr[11])
.channel_type = to_string(arr[12])
.device_type = to_string(arr[13])
.session_token_a = to_string(arr[14])
.session_token_b = to_string(arr[15])
.session_token_c = to_string(arr[16])
.customer_internal_id = to_string(arr[17])
.customer_external_id = to_string(arr[18])
.counterparty_id_a = to_string(arr[19])
.counterparty_id_b = to_string(arr[20])
.auth_context_start = to_string(arr[21])
.auth_context_end = to_string(arr[22])


.auth_context_start = string(.auth_context_start)
arr1 = split(.auth_context_start, ",")

.auth_channel_start = to_string(arr1[0])
.entrypoint_node_start = to_string(arr1[1])
.entry_time_start, err = if arr1[2] != "" {
    unix_timestamp = to_unix_timestamp(parse_timestamp!(arr1[2], "%+"))
    unix_timestamp = to_int(unix_timestamp) + 10800
    format_timestamp(from_unix_timestamp!(unix_timestamp), "%Y-%m-%d %H:%M:%S")
} else {
    ""
}


.auth_context_end = string(.auth_context_end)
arr2 = split(.auth_context_end, ",")

.auth_channel_end = to_string(arr2[0])
.entrypoint_node_end = to_string(arr2[1])
.entry_time_end, err = if arr2[2] != "" {
    unix_timestamp = to_unix_timestamp(parse_timestamp!(arr2[2], "%+"))
    unix_timestamp = to_int(unix_timestamp) + 10800
    format_timestamp(from_unix_timestamp!(unix_timestamp), "%Y-%m-%d %H:%M:%S")
} else {
     ""
}


.compliance_rules = replace(.compliance_rules, ";", "&")

.compliance_rules, err = replace_with(.compliance_rules, r'\[(.*?)\]') -> |m| {
    "[" + replace(m.captures[0], ",", "&") + "]"
}

.used_services_raw = string(.used_services_raw)
.services_info = split(.used_services_raw, ";")
.parsed_services_info = []

for_each(.services_info) -> |_index, value| {
    next_service = split(value, ",")
        
    service_call_time, err = if next_service[5] != "" {
        unix_timestamp = to_unix_timestamp(parse_timestamp!(next_service[5], "%+"))
        unix_timestamp = to_int(unix_timestamp) + 10800
        format_timestamp(from_unix_timestamp!(unix_timestamp), "%Y-%m-%d %H:%M:%S")
    } else {
        ""
    }
            
    obj = {
        "service_name": to_string(next_service[0]),
        "service_mode": to_string(next_service[1]),
        "number_of_diversions": to_string(next_service[2]),
        "assoc_party_id": to_string(next_service[3]),
        "service_id": to_string(next_service[4]),
        "service_call_time": service_call_time,
        "number_of_participants": to_string(next_service[6]),
        "action_type": to_string(next_service[7]),
        "customer_group_tag": to_string(next_service[8]),
        "service_cost_info": to_string(next_service[9])
    }
            
    .parsed_services_info = push(.parsed_services_info, obj)
}


.part2 = string(.part2)
arr3 = split(.part2, ";")
.transaction_status_code = to_string(arr3[0])
.status_http = to_string(arr3[1])
.processing_time_sec = to_string(arr3[2])
.user_interaction_time_sec = to_string(arr3[3])


.message =      .transaction_end_time + ";" +
        .transaction_type + ";" +
        .branch_code + ";" +
        .transaction_id + ";" +
        .customer_id + ";" +
        .transaction_start_time + ";" +
        .authorization_time + ";" +
        .operation_code + ";" +
        .account_sender + ";" +
        .account_receiver + ";" +
        .forwarding + ";" +
        .forwarding_reason + ";" +
        .channel_type + ";" +
        .device_type + ";" +
        .session_token_a + ";" +
        .session_token_b + ";" +
        .session_token_c + ";" +
        .customer_internal_id + ";" +
        .customer_external_id + ";" +
        .counterparty_id_a + ";" +
        .counterparty_id_b + ";" +
        .auth_channel_start + "," +
        .entrypoint_node_start + "," +
        .entry_time_start + ";" +
        .auth_channel_end + "," +
        .entrypoint_node_end + "," +
        .entry_time_end + ";\"" + 
        .security_flag + "\";" +
        .processing_code + ";\"" +
        .compliance_rules + "\";\""

for_each(.parsed_services_info) -> |index, obj| {
    if index != 0 {
        .message = .message + "&"
    }
    .message = .message + 
               obj.service_name + "," +
               obj.service_mode + "," +
               obj.number_of_diversions + "," +
               obj.assoc_party_id + "," +
               obj.service_id + "," +
               obj.service_call_time + "," +
               obj.number_of_participants + "," +
               obj.action_type + "," +
               obj.customer_group_tag + "," +
               obj.service_cost_info
        }


.message = .message + "\";" +
                .transaction_status_code + ";" +
                .status_http + ";" +
                .processing_time_sec + ";" +
                .user_interaction_time_sec + ";"


del(.host)
del(.file)
del(.timestamp)
del(.source_type)

И на выходе мы получаем такую обработанную строку:

2025-02-14 10:11:17;522;123456789101;51ab65e0-0805-498f-8508-3cc4cd459ffd;234567891012;2025-01-21 08:10:22;2025-01-21 08:10:35;234567891012;750;234567891012;;;3;1;jrccpjjjqhkcj2ar2k0srqrqa20qanc2@1.2.3.4;678F728F12B9D00030BA8@2.3.4.5:1234;;;436046005035947;;urn:86661307-056805-0;PS,43604000B000DA02,2025-02-14 10:10:28;PS,43604000B000DA02,2025-02-14 10:10:28;"";900;"345678912101,91,[],[],,AC";"OIP,override,0,,,2025-02-14 10:10:23,,,,[]&CAT,,0,,,2025-02-14 10:10:31,,,,[]";16;;55;43;

Заключение

Надеюсь, мне удалось вам донести принцип работы с VRL. Мы рассмотрели синтаксис и базовые функции для работы с данными. А также попробовали применить полученные знания на практике. Не стоит сильно заострят внимание на требованиях к обработке данных из примера, поскольку главной задачей было показать базовый подход к обработке данных, а также зацепить нестандартные методы.
Для будет приятна, если эта статья может вам освоить такой замечательный инструмент как vector и поможем вам составлять пайплайны для обработки своих данных.

VRL — просто, подробно и понятно