Чтение сообщения JMS как json, а не текста с помощью коннектора источника Kafka ActiveMQ

Добро пожаловать, Elen1no1Yami!

Мне кажется, проблема в том, что поле text сообщения представляет собой строку, содержащую полезную нагрузку JSON, которая вас интересует, но эта полезная нагрузка имеет двойные кавычки, экранированные с помощью \ char.

Я предполагаю, что данные в самом ActiveMQ не имеют символа \, но было бы хорошо, если бы вы могли это прояснить.

Я вижу подходы к решению этой проблемы:

1. иметь возможность настроить коннектор, чтобы НЕ экранировать кавычки в полезной нагрузке. Чтобы сообщение выглядело больше так:

{
  "messageID": "ID:plato-46377-1596636746117-4:4:1:1:1",
  "text": {
   "widget": {
    "debug": "on",
    "window": {
      "title": "Sample Konfabulator Widget",
      "name": "main_window",
      "width": 500,
      "height": 500
    },
    "image": {
     "src": "Images/Sun.png",
     "name": "sun1",
     "hOffset": 250,
     "vOffset": 250,
     "alignment": 
     "center"
    },
    ... etc
}

2. или каким-то образом попросите ksqlDB обработать сообщение, поскольку оно все еще имеет доступ к JSON в поле text.

Обобщает ли это то, что вы ищете? Если да, обновите свой вопрос, чтобы отразить это. (Хорошо включать такие подробности в свой вопрос, чтобы было понятно, о чем вы спрашиваете.

Что до ответа ...

1. Я не эксперт по подключению, поэтому не могу комментировать и не вижу ничего в деталях конфигурация коннектора, которая может позволить вам изменять содержимое text. Другие, кто знает больше о Connect, могут больше помочь.

2. Чтобы получить доступ к встроенному / экранированному JSON в ksqlDB, вам сначала нужно удалить экранирование. См. Ниже способы сделать это с помощью ksqlDB.

Использование ksqlDB для доступа к экранированному JSON

Прежде чем мы сможем получить доступ к документу JSON в text, мы должны удалить экранирование.

Я могу думать о двух способах развития моей головы:

Напишите собственный UDF

Лучшим способом было бы написать пользовательский UDF unescape_json, который может удалить экранирование.

-- Import raw stream with value as simple STRING containing all the payload
CREATE STREAM RAW (
  message STRING
 ) WITH (
  KAFKA_TOPIC=<something>,
  VALUE_FORMAT='KAFKA'
 );


-- Use custom UDF to process this and write it back as a properly formatted JSON document:
CREATE STREAM JSONIFIED AS
  SELECT MY_CUSTOM_UDF(message) FROM RAW;

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

Использование REPLACE для удаления экранирования

ПРИМЕЧАНИЕ: это хрупкое решение: замена символов может совпадать и заменять то, чего не следует, в зависимости от содержания вашего сообщения!

Давайте поработаем с более простыми тестовыми данными, чтобы объяснить, что нужно, например, мы хотим преобразовать:

{
 "messageID": "ID:plato-46377-1596636746117-4:4:1:1:1",
 "text": "{\"widget\": 10}"
}

To:

{
 "messageID": "ID:plato-46377-1596636746117-4:4:1:1:1",
 "text": {"widget": 10}
}

Для этого требуются три вещи:

1. Заменить проем "text": "{ на "text": {
2. Заменить все \" на ".
3. Заменить закрытие }" на }

Мы можем использовать REPLACE. функцию для этого или REGEXP_REPLACE функция:

-- Import raw stream with value as simple STRING containing all the payload
CREATE STREAM RAW (
  message STRING
 ) WITH (
  KAFKA_TOPIC=<something>,
  VALUE_FORMAT='KAFKA'
 );


-- Use REPLACE to remove reformat:
CREATE STREAM JSONIFIED AS 
  SELECT 
    REPLACE(
      REPLACE(
        REPLACE(message, 
          '"text": "{', '"text": {'), 
          '\"', '"'), 
          '"}', '}')
  FROM RAW;

Конечно, это решение потенциально может повредить ваши данные, если они содержат какие-либо условия поиска: "text": "{, \" или "} где-либо еще в ваших данных, например

{
 "messageID": "ID:plato-46377-1596636746117-4:4:1:1:1",
 "text": "{\"widget\": \"hello \\\"} world\"}"
}

Было бы неправильно преобразовано в

{
 "messageID": "ID:plato-46377-1596636746117-4:4:1:1:1",
 "text": {"widget": "hello \\}world"}
}

Вот почему предпочтительнее использовать пользовательский UDF.

После того, как вы исправили содержимое своего ввода (и записали его в новую тему), вы можете импортировать свои данные как обычно:

CREATE STREAM DATA (
   messageId STRING,
   text STRUCT<Widget INT>
 ) WITH (
   kafka_topic='JSONIFIED',
   value_format='JSON'
 );