Создание потока KSQL: как извлечь значение из сложного json - PullRequest
Новая
       9

Создание потока KSQL: как извлечь значение из сложного json

0 голосов
/ 18 мая 2018

Я пытаюсь создать поток в Apache / KAFKA KSQL. Тема содержит (довольно сложный JSON) 

{
  "agreement_id": "dd8afdbe-59cf-4272-b640-b14a24d8234c",
  "created_at": "2018-02-17 16:00:00.000Z",
  "id": "6db276a8-2efe-4495-9908-4d3fc4cc16fa",
  "event_type": "data",
  "total_charged_amount": {
    "tax_free_amount": null,
    "tax_amounts": [],
    "tax_included_amount": {
      "amount": 0.0241,
      "currency": "EUR"
    }
  }
  "used_service_units": [
    {
      "amount": 2412739,
      "currency": null,
      "unit_of_measure": "bytes"
    }
  ]
}

Теперь создать поток легко, просто используя такие вещи, как event_type и made_at.Это было бы так:

CREATE STREAM tstream (event_type varchar, created_at varchar) WITH (kafka_topic='usage_events', value_format='json');

Но теперь мне нужно получить доступ к used_service_units .... и я хотел бы извлечь "количество" в JSON выше

Как бы я это сделал? 

CREATE STREAM usage (event_type varchar,create_at varchar, used_service_units[0].amount int) WITH (kafka_topic='usage_events', value_format='json');

Результат 

line 1:78: mismatched input '[' expecting {'ADD', 'APPROXIMATE', ...

И если я вместо этого создаю поток, подобный этому 

CREATE STREAM usage (event_type varchar,create_at varchar, used_service_units varchar) WITH (kafka_topic='usage_events', value_format='json');

И затемSQL SELECT в потоке, как это 

SELECT EXTRACTJSONFIELD(used_service_units,'$.amount') FROM usage;
SELECT EXTRACTJSONFIELD(used_service_units[0],'$.amount') FROM usage;
SELECT EXTRACTJSONFIELD(used_service_units,'$[0].amount') FROM usage;

Ни один из этих вариантов не работает ...

Этот дал мне 

SELECT EXTRACTJSONFIELD(used_service_units[0],'$.amount') FROM usage;'

Code generation failed for SelectValueMapper

1 Ответ

0 голосов
/ 18 мая 2018

Похоже, что ОДИН способ решения этой проблемы - сделать тип данных столбца массивом, т.е. 

CREATE STREAM usage (event_type varchar,created_at varchar, total_charged_amount varchar, used_service_units array<varchar> ) WITH (kafka_topic='usage_events', value_format='json');

Теперь я могу сделать следующее: 

SELECT EXTRACTJSONFIELD(used_service_units[0],'$.amount') FROM usage
...