PipelineDB consumer from KAFKA JSON with array

Question

My kafka send the following json

'{
"eventSummaryList": [
    {
        "customer": 1,
        "data": "{\"cliente\":\"52264\",\"data_posicao\":\"1484250682\",\"gps_valido\":\"1\",\"horimetro\":\"0\",\"ibuttonPart1\":\"0\",\"ibuttonPart2\":\"0\",\"id_evento\":\"null\",\"id_motorista\":\"0\",\"ignicao\":\"0\",\"latitude\":\"-25.5385123\",\"longitude\":\"-49.1995068\",\"odometro\":\"0\",\"pos_memoria\":\"0\",\"veiculo\":\"103970\",\"velocidade\":\"0\"}",
        "identifierRule": 1770,
        "identifierSummary": 17,
        "rule": "rota_fora",
        "status": 1,
        "vehicle": 103970
    },
    {
        "customer": 2,
        "data": "{\"cliente\":\"52264\",\"data_posicao\":\"1484250682\",\"gps_valido\":\"1\",\"horimetro\":\"0\",\"ibuttonPart1\":\"0\",\"ibuttonPart2\":\"0\",\"id_evento\":\"null\",\"id_motorista\":\"0\",\"ignicao\":\"0\",\"latitude\":\"-25.5385123\",\"longitude\":\"-49.1995068\",\"odometro\":\"0\",\"pos_memoria\":\"0\",\"veiculo\":\"103970\",\"velocidade\":\"0\"}",
        "identifierRule": 8,
        "identifierSummary": 7,
        "rule": "velocidade_maior",
        "status": 1,
        "vehicle": 103970
    }
]

}'

I created this continuous transfomr

CREATE CONTINUOUS TRANSFORM sensor_event_process_transform AS 

                 SELECT   cast ( cast(pack ->>'eventSummaryList' as json)->>'customer'  as bigint ) as customer
                 FROM   pipeline_kafka.sensor_event_process_stream  

THEN EXECUTE procedure update_sensor_event_process_t();

but my log pipelineDB return this...

CONTEXT: JSON data, line 1: { COPY sensor_event_process_stream, line 1, column pack: "{" LOG: [pipeline_kafka] sensor_event_process_stream <- topicNotificationProcess (PID 25201): failed to process batch, dropped 8 messages ERROR: invalid input syntax for type json DETAIL: The input string ended unexpectedly.

How do I go through the json array and get only the contents of the customer column?

Answer 1

hi dears i solved my problem, i user the function json_array_elements, stayed like this ...

  CREATE CONTINUOUS TRANSFORM sensor_event_process_transform AS 
SELECT Cast(value::json ->> 'identifierRule' AS               BIGINT) AS id_regra, 
       Cast(value::json ->> 'rule' AS                         VARCHAR) AS regra, 
       Cast((value::json ->'data')::json->> 'veiculo' AS      BIGINT) AS id_veiculo,
       Cast(value::json ->> 'customer' AS bigint) AS id_cliente , 
       cast((value::json ->'data')::json->> 'velocidade' AS   int)    AS velocidade, 
       cast((value::json ->'data')::json->> 'odometro' AS     int)    AS odometro, 
       cast((value::json ->'data')::json->> 'data_posicao' AS bigint) AS data_posicao, 
       cast((value::json ->'data')::json->> 'id_motorista' AS bigint) AS id_motorista, 
       cast((value::json ->'data')::json->> 'latitude' AS float)      AS latitude, 
       cast((value::json ->'data')::json->> 'longitude' AS float)     AS longitude, 
       cast(value::json ->> 'status' AS boolean)                      AS status 
FROM   ( 
              SELECT (json_array_elements(pack->'eventSummaryList'))::json AS value 
              FROM   pipeline_kafka.sensor_event_process_stream ) b
THEN EXECUTE procedure update_sensor_event_process_t();

An important detail, do not pass json with spaces and line breaks, the pipeline doesn't accpet.