ksqldb: развертывание нескольких элементов списка

На чтение 2 мин Опубликовано 18.09.2024

Вопрос или проблема

У меня есть данные, такие как:

Одно сообщение Kafka:

[{"source": "858256_6052+571", "numericValue": null, "created": 1725969039288, "textValue": "mytestData"}]

Еще одно сообщение Kafka:

[{"source": "858256_6052+571", "numericValue": null, "created": 1725969039288, "textValue": "mytestData2"}, {"source": "858256_6052+571", "numericValue": null, "created": 1725969039288, "textValue": "mytestData3"}]

Я хочу, чтобы ksql создал поток из трех элементов для этого. Как мне это сделать? Я уже пробовал использовать explode, но для этого мне нужен ключ, которого нет в моих данных.

Ответ или решение

Чтобы решить вашу задачу с использованием ksqlDB и произвести "взрыв" (explode) элементов вашего JSON-массива, вам нужно будет создать поток, который будет извлекать элементы из массива и представлять каждый объект массива в отдельной записи.

Вот шаги, которые нужно выполнить:

Создайте стрим, если у вас его еще нет. Сначала вам необходимо создать поток из ваших данных. Предположим, что у вас есть Kafka-топик, содержащий ваши сообщения (например, your_topic_name):

CREATE STREAM your_stream_name (
    source VARCHAR,
    numericValue DOUBLE,
    created BIGINT,
    textValue VARCHAR
) WITH (
    kafka_topic='your_topic_name',
    value_format='JSON'
);

Используйте функцию EXPLODE для разбивания массива JSON. В ksqlDB функция EXPLODE позволяет извлекать элементы из массива и создавать новый поток. Однако вам нужно будет обрабатывать ваши данные в виде массива. Можно использовать ARRAY в вашем потоке и затем применить EXPLODE:

CREATE STREAM exploded_stream AS
SELECT
    source,
    numericValue,
    created,
    textValue
FROM your_stream_name
LATERAL VIEW EXPLODE(CAST(JSON_PARSE(textValue) AS ARRAY<STRUCT<source VARCHAR, numericValue DOUBLE, created BIGINT, textValue VARCHAR>>)) AS exploded_element
EMIT CHANGES;

Вывод данных. После выполнения предыдущего запроса в exploded_stream будут собраны все ваши элементы в отдельные записи. Теперь вы можете выполнять дальнейшие операции с полученными данными, или просто вывести их:

SELECT * FROM exploded_stream EMIT CHANGES;

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

Обратите внимание, что в примере был использован JSON_PARSE, который может потребовать корректировки в зависимости от формата ваших исходных данных. Если надо использовать другое представление данных или формат сообщения, стоит внести изменения соответствующим образом.

В заключение, используя EXPLODE в сочетании с правильной структурой вашего потока, вы можете эффективно преобразовывать массивы JSON в плоские записи в ksqlDB.