DataWorks:LogHub (SLS)

Recursos

Limitações

Durante a gravação offline no LogHub (SLS) pelo Data Integration, pode ocorrer duplicação de dados se uma tarefa for reexecutada após um failover, pois o LogHub (SLS) não é idempotente.

Tipos de campo suportados

O Data Integration suporta a leitura e a gravação dos seguintes tipos de campo do LogHub (SLS).

Detalhes:

• Ao gravar dados no LogHub (SLS) no modo offline

  Todos os tipos de dados suportados são convertidos para STRING antes da gravação no LogHub (SLS). A tabela a seguir lista as conversões de tipo de dados para o writer do LogHub (SLS).

  Tipo interno do Data Integration suportado	Tipo de dados ao gravar no LogHub (SLS)
  LONG	STRING
  DOUBLE	STRING
  STRING	STRING
  DATE	STRING
  BOOLEAN	STRING
  BYTES	STRING

• Ao ler dados do LogHub (SLS) no modo em tempo real

  Os seguintes campos de metadados são incluídos por padrão.

  Campo de sincronização em tempo real do LogHub (SLS)	Tipo de dados	Descrição
  __time__	STRING	Campo reservado do SLS: __time__. Horário do log especificado durante a gravação dos dados. Trata-se de um timestamp UNIX em segundos.
  __source__	STRING	Campo reservado do SLS: __source__. Dispositivo de origem do log.
  __topic__	STRING	Campo reservado do SLS: __topic__. Nome do tópico.
  __tag__:__receive_time__	STRING	Horário de chegada do log ao servidor. Se o recurso de registro de endereços IP públicos estiver ativado, o servidor anexa este campo ao log bruto após o recebimento. Trata-se de um timestamp UNIX em segundos.
  __tag__:__client_ip__	STRING	Endereço IP público do dispositivo de origem do log. Se o recurso de registro de endereços IP públicos estiver ativado, o servidor anexa este campo ao log bruto após o recebimento.
  __tag__:__path__	STRING	Caminho do arquivo de log coletado pelo Logtail. O Logtail anexa automaticamente este campo ao log.
  __tag__:__hostname__	STRING	Nome da máquina da qual o Logtail coleta dados. O Logtail anexa automaticamente este campo ao log.

Criar uma fonte de dados

Desenvolver uma tarefa de sincronização de dados

Para obter informações sobre o ponto de entrada e o procedimento de configuração de uma tarefa de sincronização, consulte os guias a seguir.

Perguntas frequentes

• Dados existem em um campo de origem do LogHub, mas estão vazios após a sincronização.

• Dados ausentes após a sincronização do LogHub.

• Os campos lidos do LogHub durante o mapeamento de campos não correspondem ao esperado.

Para mais perguntas frequentes sobre o Data Integration, consulte Perguntas frequentes sobre o Data Integration.

Apêndice 1: Exemplos de script e descrições de parâmetros

{
 "type":"job",
 "version":"2.0",// Version number.
 "steps":[
     {
         "stepType":"LogHub",// Plug-in name.
         "parameter":{
             "datasource":"",// Data source.
             "column":[// Fields.
                 "col0",
                 "col1",
                 "col2",
                 "col3",
                 "col4",
                 "C_Category",
                 "C_Source",
                 "C_Topic",
                 "C_MachineUUID", // Topic.
                 "C_HostName", // Hostname.
                 "C_Path", // Path.
                 "C_LogTime" // Event time.
             ],
             "beginDateTime":"",// The start time for data consumption.
             "batchSize":"",// The number of data entries to query from Simple Log Service at a time.
             "endDateTime":"",// The end time for data consumption.
             "fieldDelimiter":",",// Column delimiter.
             "logstore":""// The name of the destination Logstore.
         },
         "name":"Reader",
         "category":"reader"
     },
     { 
         "stepType":"stream",
         "parameter":{},
         "name":"Writer",
         "category":"writer"
     }
 ],
 "setting":{
     "errorLimit":{
         "record":"0"// The number of error records.
     },
     "speed":{
         "throttle":true,// If throttle is set to false, the mbps parameter does not take effect, and the data rate is not limited. If throttle is set to true, the data rate is limited.
            "concurrent":1, // The number of concurrent jobs.
            "mbps":"12"// The maximum data rate. 1 mbps = 1 MB/s.
     }
 },
 "order":{
     "hops":[
         {
             "from":"Reader",
             "to":"Writer"
         }
     ]
 }
}

{
    "type": "job",
    "version": "2.0",// Version number.
    "steps": [
        { 
            "stepType": "stream",
            "parameter": {},
            "name": "Reader",
            "category": "reader"
        },
        {
            "stepType": "LogHub",// Plug-in name.
            "parameter": {
                "datasource": "",// Data source.
                "column": [// Fields.
                    "col0",
                    "col1",
                    "col2",
                    "col3",
                    "col4",
                    "col5"
                ],
                "topic": "",// Select a topic.
                "batchSize": "1024",// The number of records in a batch submission.
                "logstore": ""// The name of the destination Simple Log Service Logstore.
            },
            "name": "Writer",
            "category": "writer"
        }
    ],
    "setting": {
        "errorLimit": {
            "record": ""// The number of error records.
        },
        "speed": {
            "throttle":true,// If throttle is set to false, the mbps parameter does not take effect, and the data rate is not limited. If throttle is set to true, the data rate is limited.
            "concurrent":3, // The number of concurrent jobs.
            "mbps":"12"// The maximum data rate. 1 mbps = 1 MB/s.
        }
    },
    "order": {
        "hops": [
            {
                "from": "Reader",
                "to": "Writer"
            }
        ]
    }
}

Apêndice 2: Sintaxe SPL para filtragem

Quando o LogHub for usado como origem, filtre dados usando a sintaxe de consulta do LogHub ou instruções Structured Process Language (SPL). A tabela a seguir descreve a sintaxe.

SELECT * WHERE Type='write'

SELECT "__tag__:node" AS node, path

SELECT 
  CAST(Status AS BIGINT) AS Status, 
  date_parse(Time, '%Y-%m-%d %H:%i') AS Time

| extend Status=cast(Status as BIGINT), extend Time=date_parse(Time, '%Y-%m-%d %H:%i')

SELECT 
  CAST(Status AS BIGINT) AS Status, 
  date_parse(Time, '%Y-%m-%d %H:%i') AS Time

SELECT 
      CAST(Status AS BIGINT) AS Status, 
      date_parse(Time, '%Y-%m-%d %H:%i') AS Time