Todos os produtos
Search

Este produto

    Simple Log Service:Transformar dados JSON complexos

    Central de documentação

    Simple Log Service:Transformar dados JSON complexos

    Última atualização: Jul 03, 2026

    Este documento descreve como usar o recurso de transformação de dados do Simple Log Service (SLS) para transformar dados JSON complexos.

    Transformar dados JSON complexos com múltiplas subchaves como arrays

    Logs de programas frequentemente adotam um formato JSON estatístico. Esses registros geralmente contêm informações básicas e várias subchaves estruturadas como arrays. Por exemplo, um servidor pode gerar um log a cada minuto com o status atual e dados estatísticos sobre os nós relacionados do servidor e do cliente.

    • Log de exemplo

      __source__:  192.0.2.1
__topic__:  
content:{
     "service": "search_service",
     "overal_status": "yellow",
     "servers": [
         {
             "host": "192.0.2.1",
             "status": "green"
         },
         {
             "host": "192.0.2.2",
             "status": "green"
         }
     ],
     "clients": [
         {
             "host": "192.0.2.3",
             "status": "green"
         },
         {
             "host": "192.0.2.4",
             "status": "red"
         }
     ]
}

    • Requisitos de transformação de dados

      1. Divida o log bruto em três logs distintos com base no campo topic: overall_type, client_status e server_status.

      2. Armazene informações específicas para cada valor de topic.

        • overall_type: Mantenha a contagem de servidores, a contagem de clientes, a cor do overall_status e as informações do serviço.

        • client_status: Mantenha o endereço do host, o status e as informações do serviço.

        • server_status: Mantenha o endereço do host, o status e as informações do serviço.

    • Resultado esperado

      __source__:  192.0.2.1
__topic__:  overall_type
client_count:  2
overal_status:  yellow
server_count:  2
service:  search_service

__source__:  192.0.2.1
__topic__:  client_status
host:  192.0.2.4
status:  red
service:  search_service

__source__:  192.0.2.1
__topic__:  client_status
host:  192.0.2.3
status:  green
service:  search_service

__source__:  192.0.2.1
__topic__:  server_status
host:  192.0.2.1
status:  green
service:  search_service

__source__:  192.0.2.1
__topic__:  server_status
host:  192.0.2.2
status:  green
service:  search_service

    • Solução

      1. Divida o log em três registros separados. Para isso, atribua três valores diferentes ao campo topic. Após a divisão, haverá três logs idênticos, exceto pelo campo topic.

        e_set("__topic__", "server_status,client_status,overall_type")
e_split("__topic__")

        O formato do log após o processamento é:

        __source__:  192.0.2.1
__topic__:  server_status         // The other two logs have `client_status` and `overall_type` as topics. The rest of the fields are the same.
content:  {
    ...As before...
}

      2. Expanda o conteúdo JSON de primeiro nível do campo content e remova esse campo em seguida.

        e_json('content',depth=1)
e_drop_fields("content")

        O formato do log após o processamento é:

        __source__:  192.0.2.1
__topic__:  overall_type              // The other two logs have `client_status` and `overall_type` as topics. The rest of the fields are the same.
clients:  [{"host": "192.0.2.3", "status": "green"}, {"host": "192.0.2.4", "status": "red"}]
overal_status:  yellow
servers:  [{"host": "192.0.2.1", "status": "green"}, {"host": "192.0.2.2", "status": "green"}]
service:  search_service

      3. Para o log com o tópico overall_type, calcule os valores de client_count e server_count.

        e_if(e_search("__topic__==overall_type"), 
     e_compose(
        e_set("client_count", json_select(v("clients"), "length([*])", default=0)), 
        e_set("server_count", json_select(v("servers"), "length([*])", default=0))
  ))

        O log processado resulta em:

        __topic__:  overall_type
server_count:  2
client_count:  2

      4. Remova os campos desnecessários.

        e_if(e_search("__topic__==overall_type"), e_drop_fields("clients", "servers"))

      5. Divida novamente o log com o tópico server_status.

        e_if(e_search("__topic__==server_status"), 
     e_compose(
        e_split("servers"), 
        e_json("servers", depth=1)
  ))

        O log divide-se nos dois registros seguintes:

        __topic__:  server_status
servers:  {"host": "192.0.2.1", "status": "green"}
host: 192.0.2.1
status: green
        __topic__:  server_status
servers:  {"host": "192.0.2.2", "status": "green"}
host: 192.0.2.2
status: green

      6. Mantenha apenas os campos relevantes:

        e_if(e_search("__topic__==overall_type"), e_drop_fields("servers"))

      7. Divida novamente o log com o tópico client_status e exclua o campo clients.

        e_if(e_search("__topic__==client_status"), 
     e_compose(
        e_split("clients"), 
        e_json("clients", depth=1),
        e_drop_fields("clients")
  ))

        O log divide-se nos dois registros seguintes:

        __topic__:  client_status
host: 192.0.2.3
status: green
        __topic__:  clients
host: 192.0.2.4
status: red

      8. As regras completas da linguagem específica de domínio (DSL) de LOG são:

        # Split the log.
e_set("__topic__", "server_status,client_status,overall_type")
e_split("__topic__")
e_json('content',depth=1)
e_drop_fields("content")

# Process the overall_type log.
e_if(e_search("__topic__==overall_type"), 
     e_compose(
        e_set("client_count", json_select(v("clients"), "length([*])", default=0)),
				e_set("server_count", json_select(v("servers"), "length([*])", default=0))
))

# Process the server_status log.
e_if(e_search("__topic__==server_status"), 
     e_compose(
        e_split("servers"), 
        e_json("servers", depth=1)
  ))
e_if(e_search("__topic__==overall_type"), e_drop_fields("servers"))

# Process the client_status log.
e_if(e_search("__topic__==client_status"), 
     e_compose(
        e_split("clients"), 
        e_json("clients", depth=1),
        e_drop_fields("clients")
  ))

    Otimização da solução

    A solução anterior apresenta problemas quando content.servers e content.servers estão vazios. Considere, por exemplo, o seguinte log bruto:

    __source__:  192.0.2.1
__topic__:  
content:{
            "service": "search_service",
            "overal_status": "yellow",
            "servers": [ ],
            "clients": [ ]
}

    Se você usar a solução anterior para dividir esse log bruto em três partes, os logs com os tópicos client_status e server_status ficarão vazios.

    __source__:  192.0.2.1
__topic__:  overall_type
client_count:  0
overal_status:  yellow
server_count:  0
service:  search_service

__source__:  192.0.2.1
__topic__:  client_status
service:  search_service
__source__:  192.0.2.1

__topic__:  server_status
host:  192.0.2.1
status:  green
service:  search_service

    • Solução 1

      Após a divisão inicial, verifique se os logs com os tópicos server_status e client_status estão vazios. Em caso afirmativo, descarte-os.

      # For server_status: discard if empty, retain if not.
e_keep(op_and(e_search("__topic__==server_status"), json_select(v("servers"), "length([*])")))

# For client_status: discard if empty, retain if not.
e_keep(op_and(e_search("__topic__==client_status"), json_select(v("clients"), "length([*])")))

      As regras completas da LOG DSL são:

      # Split the log.
e_set("__topic__", "server_status,client_status,overall_type")
e_split("__topic__")
e_json('content',depth=1)
e_drop_fields("content")

# Process the overall_type log.
e_if(e_search("__topic__==overall_type"), 
     e_compose(
       e_set("client_count", json_select(v("clients"), "length([*])", default=0)),
			 e_set("server_count", json_select(v("servers"), "length([*])", default=0))
))

# New: Pre-process server_status: discard if empty, retain if not.
e_keep(op_and(e_search("__topic__==server_status"), json_select(v("servers"), "length([*])")))

# Process the server_status log.
e_if(e_search("__topic__==server_status"), 
     e_compose(
        e_split("servers"), 
        e_json("servers", depth=1)
  ))
e_if(e_search("__topic__==overall_type"), e_drop_fields("servers"))

# New: Pre-process client_status: discard if empty, retain if not.
e_keep(op_and(e_search("__topic__==client_status"), json_select(v("clients"), "length([*])")))

# Process the client_status log.
e_if(e_search("__topic__==client_status"), 
     e_compose(
        e_split("clients"), 
        e_json("clients", depth=1),
        e_drop_fields("clients")
  ))

    • Solução 2

      Verifique se um campo está vazio antes de dividir o log. Se o campo não estiver vazio, divida o log com base nele.

      # Set the initial topic.
e_set("__topic__", "server_status")

# If the content.servers field is not empty, split the log to create a log with the topic server_status.
e_if(json_select(v("content"), "length(servers[*])"),
   e_compose(
      e_set("__topic__", "server_status,overall_type"),
      e_split("__topic__")
   ))

# If the content.clients field is not empty, further split the log to create a log with the topic client_status.
e_if(op_and(e_search("__topic__==overall_type"), json_select(v("content"), "length(clients[*])")),
   e_compose(
      e_set("__topic__", "client_status,overall_type"),
      e_split("__topic__")
   ))

      As regras completas da LOG DSL são:

      # Split the log.
e_set("__topic__", "server_status")

# If the content.servers field is not empty, split the log to create a log with the topic server_status.
e_if(json_select(v("content"), "length(servers[*])"),
   e_compose(
      e_set("__topic__", "server_status,overall_type"),
      e_split("__topic__")
   ))

# If the content.clients field is not empty, further split the log to create a log with the topic client_status.
e_if(op_and(e_search("__topic__==overall_type"), json_select(v("content"), "length(clients[*])")),
   e_compose(
      e_set("__topic__", "client_status,overall_type"),
      e_split("__topic__")
   ))

# Process the overall_type log.
e_if(e_search("__topic__==overall_type"), 
     e_compose(
        e_set("client_count", json_select(v("clients"), "length([*])", default=0)),
				e_set("server_count", json_select(v("servers"), "length([*])", default=0))
))

# Process the server_status log.
e_if(e_search("__topic__==server_status"), 
     e_compose(
        e_split("servers"), 
        e_json("servers", depth=1)
  ))
e_if(e_search("__topic__==overall_type"), e_drop_fields("servers"))

# Process the client_status log.
e_if(e_search("__topic__==client_status"), 
     e_compose(
        e_split("clients"), 
        e_json("clients", depth=1),
        e_drop_fields("clients")
  ))

    Comparação das soluções

    • A Solução 1 apresenta redundância lógica porque gera logs vazios a partir do registro bruto e depois os exclui. No entanto, as regras são simples e fáceis de manter. Esta é a abordagem padrão recomendada.

    • A Solução 2 é mais eficiente porque verifica se há campos vazios antes da divisão. Contudo, as regras são um pouco mais complexas. Recomenda-se esta opção apenas para cenários específicos, como quando a divisão inicial pode gerar muitos eventos extras.

    Transformar dados JSON complexos com objetos de array aninhados em múltiplas camadas

    Este exemplo mostra como processar um objeto complexo que contém arrays aninhados em várias camadas. O objetivo é separar cada evento de login presente no array login_histories de cada objeto dentro do array users em um log de evento de login independente.

    • Log bruto

      __source__:  192.0.2.1
__topic__:  
content:{
  "users": [
    {
        "name": "user1",
        "login_histories": [
          {
            "date": "2019-10-10 0:0:0",
            "login_ip": "192.0.2.6"
          },
          {
            "date": "2019-10-10 1:0:0",
            "login_ip": "192.0.2.6"
          },
      {
      ...More logon information...
      }
        ]
    },
    {
        "name": "user2",
        "login_histories": [
          {
            "date": "2019-10-11 0:0:0",
            "login_ip": "192.0.2.7"
          },
          {
            "date": "2019-10-11 1:0:0",
            "login_ip": "192.0.2.9"
          },
      {
      ...More logon information...
      }     
        ]
    },
  {
    ...More users...
  }
  ]
}

    • Logs esperados após a divisão

      __source__:  192.0.2.1
name:  user1
date:  2019-10-11 1:0:0
login_ip:  192.0.2.6

__source__: 192.0.2.1
name:  user1
date:  2019-10-11 0:0:0
login_ip:  192.0.2.6

__source__:  192.0.2.1
name:  user2
date:  2019-10-11 0:0:0
login_ip:  192.0.2.7

__source__: 192.0.2.1
name:  user2
date:  2019-10-11 1:0:0
login_ip:  192.0.2.9  

...More logs...

    • Solução

      1. Divida e expanda o log com base em users no campo content.

        e_split("content", jmes='users[*]', output='item')
e_json("item",depth=1)

        Os logs processados ficam assim:

        __source__:  192.0.2.1
__topic__:  
content:{...Same as that in the raw log...}
item:  {"name": "user1", "login_histories": [{"date": "2019-10-10 0:0:0", "login_ip": "192.0.2.6"}, {"date": "2019-10-10 1:0:0", "login_ip": "192.0.2.6"}]}
login_histories:  [{"date": "2019-10-10 0:0:0", "login_ip": "192.0.2.6"}, {"date": "2019-10-10 1:0:0", "login_ip": "192.0.2.6"}]
name:  user1

__source__:  192.0.2.1
__topic__:  
content:{...Same as that in the raw log...}
item:  {"name": "user2", "login_histories": [{"date": "2019-10-11 0:0:0", "login_ip": "192.0.2.7"}, {"date": "2019-10-11 1:0:0", "login_ip": "192.0.2.9"}]}
login_histories:  [{"date": "2019-10-11 0:0:0", "login_ip": "192.0.2.7"}, {"date": "2019-10-11 1:0:0", "login_ip": "192.0.2.9"}]
name:  user2

      2. Em seguida, divida e expanda os dados com base em login_histories.

        e_split("login_histories")
e_json("login_histories", depth=1)

        Os logs processados ficam assim:

        __source__:  192.0.2.1
__topic__: 
content: {...Same as that in the raw log...}
date:  2019-10-11 0:0:0
item:  {"name": "user2", "login_histories": [{"date": "2019-10-11 0:0:0", "login_ip": "192.0.2.7"}, {"date": "2019-10-11 1:0:0", "login_ip": "192.0.2.9"}]}
login_histories:  {"date": "2019-10-11 0:0:0", "login_ip": "192.0.2.7"}
login_ip:  192.0.2.7
name:  user2

__source__:  192.0.2.1
__topic__: 
content: {...Same as that in the raw log...}
date:  2019-10-11 1:0:0
item:  {"name": "user2", "login_histories": [{"date": "2019-10-11 0:0:0", "login_ip": "192.0.2.7"}, {"date": "2019-10-11 1:0:0", "login_ip": "192.0.2.9"}]}
login_histories:  {"date": "2019-10-11 1:0:0", "login_ip": "192.0.2.9"}
login_ip:  192.0.2.9
name:  user2

__source__: 192.0.2.1
__topic__:  
content: {...Same as that in the raw log...}
date:  2019-10-10 1:0:0
item:  {"name": "user1", "login_histories": [{"date": "2019-10-10 0:0:0", "login_ip": "192.0.2.6"}, {"date": "2019-10-10 1:0:0", "login_ip": "192.0.2.6"}]}
login_histories:  {"date": "2019-10-10 1:0:0", "login_ip": "192.0.2.6"}
login_ip:  192.0.2.6
name:  user1

__source__: 192.0.2.1
__topic__:  
content: {...Same as that in the raw log...}
date:  2019-10-10 0:0:0
item:  {"name": "user1", "login_histories": [{"date": "2019-10-10 0:0:0", "login_ip": "192.0.2.6"}, {"date": "2019-10-10 1:0:0", "login_ip": "192.0.2.6"}]}
login_histories:  {"date": "2019-10-10 0:0:0", "login_ip": "192.0.2.6"}
login_ip:  192.0.2.6
name:  user1

      3. Por fim, remova os campos irrelevantes.

        e_drop_fields("content", "item", "login_histories")

        Os logs processados ficam assim:

        __source__: 192.0.2.1
__topic__:
name:  user1
date:  2019-10-11 1:0:0
login_ip:  192.0.2.6

__source__:  192.0.2.1
__topic__:
name:  user1
date:  2019-10-11 0:0:0
login_ip:  192.0.2.6

__source__:  192.0.2.1
__topic__:
name:  user2
date:  2019-10-11 0:0:0
login_ip:  192.0.2.7

__source__: 192.0.2.1
__topic__:
name:  user2
date:  2019-10-11 1:0:0
login_ip:  192.0.2.9

      4. As regras completas da LOG DSL podem ser escritas da seguinte forma:

        e_split("content", jmes='users[*]', output='item')
e_json("item",depth=1)
e_split("login_histories")
e_json("login_histories", depth=1)
e_drop_fields("content", "item", "login_histories")

    Resumo: Para requisitos semelhantes, divida primeiro o log, expanda os dados em seguida e, por último, exclua os campos irrelevantes.