DataWorks:FAQ sobre sincronização em lote

• Se o teste de conectividade foi bem-sucedido anteriormente, execute-o novamente para confirmar se o grupo de recursos e o banco de dados estão conectados no momento (e se nenhuma alteração foi feita no lado do banco de dados).

• Verifique se o grupo de recursos que passou no teste de conectividade é o mesmo utilizado para executar a tarefa.

  Verifique o grupo de recursos usado pela tarefa:

  • Se a tarefa for executada no grupo de recursos padrão, os logs conterão as seguintes informações: running in Pipeline[basecommon_ group_xxxxxxxxx]

  • Se a tarefa for executada em um grupo de recursos exclusivo para Data Integration, os logs conterão as seguintes informações: running in Pipeline[basecommon_S_res_group_xxx]

  • Se a tarefa for executada em um grupo de recursos serverless, os logs conterão as seguintes informações: running in Pipeline[basecommon_Serverless_res_group_xxx]

• Caso a tarefa falhe ocasionalmente durante o agendamento nas primeiras horas da manhã, mas seja bem-sucedida após uma nova execução, verifique a carga do banco de dados no momento em que o erro ocorreu.

Se uma tarefa de sincronização em lote falhar intermitentemente, a causa pode ser uma configuração incompleta da lista de permissões. Verifique se a lista de permissões do banco de dados está totalmente configurada.

Ao utilizar um grupo de recursos exclusivo para Data Integration:

• Se você adicionou anteriormente os endereços IP da interface de rede elástica (ENI) do grupo de recursos exclusivo para Data Integration à lista de permissões da fonte de dados e o grupo de recursos teve seu dimensionamento expandido desde então, atualize a lista de permissões da fonte de dados para incluir os endereços IP da ENI do grupo de recursos expandido.

• Para evitar a necessidade de atualizar a lista de permissões sempre que o grupo de recursos tiver seu dimensionamento expandido, recomendamos adicionar o bloco CIDR do vSwitch associado ao grupo de recursos exclusivo para Data Integration como a lista de permissões do banco de dados. Para mais informações, consulte Add an allowlist.

Ao utilizar um grupo de recursos serverless: consulte Network connectivity of a serverless resource group para verificar a configuração da lista de permissões do grupo de recursos e garantir que a rede esteja configurada corretamente.

Se a lista de permissões estiver configurada corretamente, verifique se a carga do banco de dados está excessivamente alta, o que pode causar interrupções nas conexões.

• Possível causa:

  A concorrência está definida com um valor muito alto, resultando em recursos insuficientes.

• Solução:

  Reduza a concorrência da tarefa de sincronização em lote.

  • Se você configurar a tarefa de sincronização em lote no modo assistente, reduza o valor de Expected Maximum Concurrency em Channel Control. Para mais informações, consulte Configure channel control in wizard mode.

  • Se você configurar a tarefa de sincronização em lote no modo script, reduza o valor do parâmetro concurrent em Channel Control. Para mais informações, consulte Configure channel control in script mode.

Para resolver este erro:

1. Se a configuração do plugin suportar parâmetros como batchsize ou maxfilesize, reduza os valores correspondentes.

   Você pode verificar se cada plugin suporta os parâmetros mencionados acima. Acesse o tópico Supported data sources and readers/writers e clique no plugin correspondente para visualizar os detalhes dos parâmetros.

2. Reduza a concorrência.

   • Se você configurar a tarefa de sincronização em lote no modo assistente, reduza o valor de Expected Maximum Concurrency em Channel Control. Para mais informações, consulte Configure channel control in wizard mode.

   • Se você configurar a tarefa de sincronização em lote no modo script, reduza o valor do parâmetro concurrent em Channel Control. Para mais informações, consulte Configure channel control in script mode.

3. Se estiver sincronizando arquivos, como arquivos do OSS, reduza o número de arquivos a serem lidos.

4. Na seção Running Resources da configuração da tarefa, aumente adequadamente o valor de Resource Usage (CU). Defina o valor de CU com cuidado para evitar afetar outras tarefas em execução.

• Mensagem de erro: Error updating database. Cause: com.mysql.jdbc.exceptions.jdbc4.MySQLIntegrityConstraintViolationException: Duplicate entry 'cfc68cd0048101467588e97e83ffd7a8-0' for key 'uk_uk_op'.

• Possível causa: O Data Integration não permite que diferentes instâncias do mesmo nó (ou seja, tarefas de sincronização com a mesma configuração JSON) sejam executadas simultaneamente. Por exemplo, se uma tarefa de sincronização for executada em um cronograma de 5 minutos e atrasos upstream fizerem com que tanto a instância das 00:00 quanto a das 00:05 sejam acionadas às 00:05, uma das instâncias não poderá ser iniciada. Isso também pode ocorrer quando você faz backfill de dados ou reexecuta uma instância enquanto a instância da tarefa ainda está em execução.

• Solução: Intercale os horários de execução das instâncias. Para tarefas agendadas em intervalos de hora ou minuto, recomendamos definir a autodependência para que a instância atual inicie somente após a conclusão da instância do ciclo anterior. Para configuração no Data Studio legado, consulte Self-dependency. Para configuração no novo Data Studio, consulte Configure self-dependency.

• Detalhes do erro: Durante uma tarefa de sincronização de dados, a tarefa falha com o seguinte erro: MongoDBReader$Task - operation exceeded time limitcom.mongodb.MongoExecutionTimeoutException: operation exceeded time limit.

• Causa possível: O volume de dados na extração completa é muito grande.

• Solução:

  • Aumente a concorrência.

  • Reduza o BatchSize.

  • Adicione a configuração cursorTimeoutInMs na seção de parâmetros do Reader e defina um valor alto, como 3600000 ms.

• Erro de leitura

  • Sintoma:

    Ao ler dados, ocorre o seguinte erro: Communications link failure The last packet successfully received from the server was 7,200,100 milliseconds ago. The last packet sent successfully to the server was 7,200,100 milliseconds ago. - com.mysql.jdbc.exceptions.jdbc4.CommunicationsException: Communications link failure

  • Causa possível:

    O banco de dados executa consultas SQL lentamente, causando timeout de leitura no MySQL.

  • Solução:

    • Verifique se há uma condição de filtro where configurada e garanta que as colunas de filtro estejam indexadas.

    • Confira se a tabela de origem contém dados em excesso. Nesse caso, divida a tarefa em várias tarefas menores.

    • Analise os logs para identificar a instrução SQL que causou o bloqueio e consulte o administrador do banco de dados para resolver o problema.

• Erro de escrita

  • Sintoma:

    Ao gravar dados, ocorre o seguinte erro: Caused by: java.util.concurrent.ExecutionException: ERR-CODE: [TDDL-4614][ERR_EXECUTE_ON_MYSQL] Error occurs when execute on GROUP 'xxx' ATOM 'dockerxxxxx_xxxx_trace_shard_xxxx': Communications link failure The last packet successfully received from the server was 12,672 milliseconds ago. The last packet sent successfully to the server was 12,013 milliseconds ago. More...

  • Causa possível:

    Uma consulta lenta causa SocketTimeout. O SocketTimeout padrão para conexões TDDL é de 12 segundos. Se uma instrução SQL levar mais de 12 segundos para executar no MySQL, o erro 4614 será reportado. Esse erro pode ocorrer ocasionalmente quando o volume de dados é grande ou o servidor está sobrecarregado.

  • Solução:

    • Aguarde a estabilização do banco de dados e execute novamente a tarefa de sincronização.

    • Entre em contato com o administrador do banco de dados para ajustar o valor de timeout.

Causa possível 2: Aguardando recursos de execução do Data Integration

Solução 2: Se os logs mostrarem status WAIT prolongado, o grupo de recursos exclusivo do Data Integration usado pela tarefa atual não possui concorrência disponível suficiente para executá-la. Para detalhes sobre a causa e a solução, consulte Troubleshoot resource group concurrency issues.

• Exemplo de cenário

  O SQL executado é o seguinte:

  SELECT bid,inviter,uid,createTime FROM `relatives` WHERE createTime>='2016-10-2300:00:00' AND reateTime<'2016-10-24 00:00:00';

  A execução iniciou em 2016-10-25 11:01:24.875 e os resultados começaram a retornar em 2016-10-25 11:11:05.489. O programa de sincronização aguardou o banco de dados retornar os resultados da consulta SQL, e o MaxCompute precisou esperar muito tempo antes de prosseguir com a execução.

• Análise da causa raiz

  A coluna createTime na cláusula WHERE não possui índice, provocando varredura completa da tabela.

• Solução

  Recomenda-se que a cláusula where utilize colunas indexadas para melhorar o desempenho. Também é possível adicionar índices conforme necessário.

Data Studio legado:

É possível modificar o grupo de recursos usado para depuração na página de detalhes da tarefa de sincronização em lote no DataStudio. Também é possível alterar o grupo de recursos de execução de tarefas do Data Integration usado durante o agendamento no Operation Center. Para mais informações, consulte Switch the Data Integration resource group.

Novo Data Studio:

É possível modificar o grupo de recursos usado para depurar tarefas do Data Integration no DataStudio. Também é possível alterar o grupo de recursos de execução de tarefas do Data Integration usado durante o agendamento no Operation Center. Para mais informações, consulte Switch the Data Integration resource group.

Dados incorretos: Registro que falha ao ser gravado no destino devido a uma exceção.

A tarefa acumula a contagem de registros de dados incorretos durante a execução. Assim que essa contagem ultrapassa o limiar configurado, a tarefa é encerrada imediatamente.

• Retenção de dados: Os dados gravados com sucesso no destino antes do término da tarefa são preservados. Nenhum rollback é realizado.

• Política de tolerância zero: Quando o limiar de dados incorretos é definido como 0, o sistema adota tolerância zero. Isso significa que a tarefa falha e para imediatamente ao detectar o primeiro registro de dado incorreto.

• Mensagem de erro:

  Se os dados contiverem caracteres emoji, erros de dados incorretos podem ocorrer durante a sincronização: [13350975-0-0-writer] ERROR StdoutPluginCollector - Dirty data {"exception":"Incorrect string value: '\\xF0\\x9F\\x98\\x82\\xE8\\xA2...' for column 'introduction' at row 1","record":[{"byteSize":8,"index":0,"rawData":9642,"type":"LONG"}],"type":"writer"} .

• Causa possível:

  • A codificação do banco de dados não está definida como utf8mb4, o que causa erros ao sincronizar caracteres emoji.

  • Os dados de origem já contêm caracteres corrompidos.

  • Há inconsistência entre a codificação do banco de dados e a do cliente.

  • A codificação do navegador é diferente, causando falhas na pré-visualização ou exibição de caracteres corrompidos.

• Solução:

  Escolha a solução adequada com base na causa dos caracteres corrompidos:

  • Se os dados originais contiverem caracteres corrompidos, corrija-os antes de executar a tarefa de sincronização.

  • Se houver inconsistência entre os formatos de codificação do banco de dados e do cliente, modifique o formato de codificação primeiro.

  • Se a codificação do navegador for diferente da codificação do banco de dados ou do cliente, unifique os formatos de codificação antes de pré-visualizar os dados.

  Você pode tentar o seguinte:

  1. Para fontes de dados adicionadas no formato JDBC, modifique para utf8mb4 da seguinte forma: jdbc:mysql://xxx.x.x.x:3306/database?com.mysql.jdbc.faultInjection.serverCharsetIndex=45.

  2. Para fontes de dados adicionadas por ID de instância, anexe o seguinte ao nome do banco de dados: database?com.mysql.jdbc.faultInjection.serverCharsetIndex=45.

  3. Modifique o formato de codificação do banco de dados para utf8mb4. Por exemplo, altere o formato de codificação do banco de dados RDS no console do RDS.

     Nota

     Comando para definir o formato de codificação da fonte de dados RDS: set names utf8mb4. Comando para verificar o formato de codificação do banco de dados RDS: show variables like 'char%'.

• Cenário: Ao sincronizar dados para o MaxCompute (ODPS), a tarefa falha ou os dados gravados são truncados porque um campo de origem excede 8 MB.

• Causa: Em tarefas de sincronização que usam MaxCompute como destino, o MaxCompute (ODPS) Writer impõe um limite de tamanho de 8 MB por campo.

• Solução:

  • Nas configurações avançadas do MaxCompute (ODPS) Writer, defina a política de tratamento de campos com tamanho excedido (overLengthRule) para especificar como lidar com campos acima do limite: truncar o campo para 8 MB, definir o campo como NULL ou gravar o campo sem truncamento.

  • Para cenários que envolvem objetos grandes, como arquivos ou logs, armazene os dados brutos no Object Storage Service (OSS) e mantenha apenas a URL no banco de dados. Em seguida, use o DataWorks para sincronizar a URL em vez do objeto grande em si.

  • Como etapa de pré-processamento, divida um campo com tamanho excedido em vários subcampos menores na origem, por exemplo, em segmentos de 7 MB. Sincronize os subcampos separadamente e concatene-os novamente no destino.

Ao criar uma tabela de destino, o DataWorks preserva apenas os nomes das colunas, tipos de dados e comentários da tabela de origem. Ele não preserva valores padrão nem restrições (incluindo restrições not-null e índices).

Tarefas de sincronização em lote não suportam o uso de chave primária composta como chave de divisão.

• Cenário: Uma tarefa de sincronização incremental falha com um erro indicando que a coluna de chave de divisão configurada (splitPk) é inválida.

• Causa possível: A configuração da chave de divisão não atende aos requisitos. Por exemplo, múltiplas colunas foram configuradas como chave de divisão, o tipo de dado da coluna configurada não é suportado ou a coluna configurada não existe na tabela.

• Solução:

  1. Atualize o mapeamento da tabela de origem para visualizar a coluna de chave de divisão sugerida automaticamente pelo sistema.

  2. Certifique-se de que apenas uma única coluna esteja configurada como chave de divisão e que seu tipo de dado seja inteiro. Conforme descrito anteriormente neste documento, o splitPk suporta apenas fragmentação de dados baseada em inteiros e não aceita strings, ponto flutuante, datas ou outros tipos.

  3. Se a coluna sugerida automaticamente não atender a essas condições, altere manualmente a chave de divisão para uma única coluna de tipo inteiro que atenda aos requisitos.

Para mais informações sobre como a chave de divisão afeta o desempenho da sincronização, consulte How do I troubleshoot a batch synchronization task that takes a long time to run?

Se ocorrerem problemas de qualidade de dados após a sincronização, consulte Troubleshoot data quality issues after synchronization para troubleshooting detalhado.

P: Como resolver o erro "Task have SSRF attacks"?

Causa: Para garantir a segurança na cloud, o DataWorks proíbe que tarefas acessem endereços internos da rede cloud por meio de endereços IP públicos. Quando uma URL na configuração do plugin (como HTTP Reader) aponta para um endereço IP interno ou nome de domínio VPC, essa verificação de segurança é acionada.

Abordagem correta:

• Não utilize endereços IP internos como 10.x.x.x ou 192.168.x.x na configuração.

• Adquira um serverless resource group (recomendado) ou um exclusive resource group for Data Integration e configure conectividade VPC para acessar fontes de dados internas com segurança.

Solução: Para tarefas que acessam fontes de dados internas, pare de usar o grupo de recursos compartilhado e migre para um serverless resource group seguro (recomendado) ou um exclusive resource group for Data Integration.

Altere a tarefa de sincronização para o modo script e adicione a seguinte configuração na seção setting da página de configuração da tarefa:

"common": {
  "column": {
    "dateFormat": "yyyyMMdd",
    "datetimeFormatInNanos": "yyyyMMdd HH:mm:ss.SSS"
  }
}

Onde:

• dateFormat especifica o formato de data usado ao converter dados do tipo DATE (sem hora) da origem para texto.

• datetimeFormatInNanos especifica o formato de data usado ao converter dados do tipo DATETIME/TIMESTAMP (com hora) da origem para texto. É possível especificar precisão até milissegundos.

1. É possível inserir constantes. Os valores devem estar entre aspas simples, como 'abc' e '123'.

2. Parâmetros de agendamento podem ser utilizados, como '${bizdate}'. Para informações sobre como usar parâmetros de agendamento, consulte Configure scheduling parameters.

3. É possível inserir as colunas de partição a serem sincronizadas, como pt.

4. Se o valor inserido não puder ser analisado, o tipo será exibido como 'Custom'.

5. Funções ODPS não são suportadas.

6. Se uma coluna adicionada manualmente for exibida como Custom (por exemplo, uma coluna de partição do MaxCompute ou uma coluna do LogHub não mostrada na pré-visualização de dados), isso não afeta a execução real da tarefa.

Na lista de mapeamento de colunas, clique em Add ou Create Field abaixo das colunas da tabela de origem, insira o nome da coluna de partição (como pt) e configure o mapeamento para a coluna da tabela de destino.

Ao configurar o MaxCompute Writer, é possível realizar operações de filtragem de colunas, reordenação e preenchimento de nulos que o próprio MaxCompute não suporta nativamente. Por exemplo, para importar todas as colunas, configure "column": ["*"].

Se a tabela MaxCompute tiver três colunas a, b e c, e você quiser sincronizar apenas as colunas c e b, configure a lista de colunas como "column": ["c","b"]. Isso significa que a primeira e a segunda colunas do Reader serão importadas para as colunas c e b da tabela MaxCompute, e a nova coluna inserida a na tabela MaxCompute será definida como nula.

Para garantir a confiabilidade da gravação de dados e evitar problemas de qualidade causados por perda de dados em colunas extras, o MaxCompute Writer reporta erro se colunas adicionais forem gravadas. Por exemplo, se a tabela MaxCompute tiver as colunas a, b e c, e o MaxCompute Writer tentar gravar mais de três colunas, um erro será reportado.

• Cenário: Uma tarefa de sincronização em lote que grava no MaxCompute (ODPS) falha, e o troubleshooting mostra que a tabela de destino contém uma coluna do tipo JSON.

• Causa possível: O MaxCompute Writer pode não suportar gravação em coluna de destino do tipo JSON em todos os casos.

• Solução: Verifique o schema da tabela MaxCompute de destino para confirmar se ela contém uma coluna do tipo JSON. Em caso afirmativo, tente um dos métodos a seguir:

  • Altere o tipo da coluna para STRING no lado do MaxCompute, por exemplo, executando ALTER TABLE ADD COLUMN ou uma alteração de schema equivalente.

  • Exclua a coluna do tipo JSON do mapeamento de campos para que ela não seja gravada durante a sincronização.

O MaxCompute Writer suporta gravação apenas na partição de último nível e não oferece roteamento de partição baseado em coluna. Se uma tabela tiver três níveis de partição, você deve especificar exatamente a partição de terceiro nível na configuração. Por exemplo, para gravar dados na partição de terceiro nível, configure como pt=20150101, type=1, biz=2. Não é possível configurar como pt=20150101, type=1 ou pt=20150101.

O MaxCompute Writer garante idempotência de gravação ao configurar "truncate": true. Quando uma gravação falha e é reexecutada, o MaxCompute Writer limpa os dados anteriores e importa novos dados, garantindo consistência após cada reexecução. Se a tarefa for interrompida devido a outras exceções durante a execução, a atomicidade dos dados não é garantida. Os dados não sofrem rollback nem reexecução automática. Utilize o recurso de idempotência para reexecutar a tarefa e garantir a integridade dos dados.

• Mensagem de erro:

  Code:DATAX_R_ODPS_005:Failed to read ODPS data, Solution:[Please contact the ODPS administrator]. RequestId=202012091137444331f60b08cda1d9, ErrorCode=StatusConflict, ErrorMessage=The download session is expired.

• Causa possível:

  Ao ler dados do MaxCompute em sincronização em lote, o sistema usa o comando tunnel do MaxCompute para upload e download de dados. Uma sessão Tunnel tem tempo de vida de 24 horas no servidor. Portanto, se uma tarefa de sincronização em lote executar por mais de 24 horas, ela falhará. Para mais informações sobre tunnel, consulte Tunnel overview.

• Solução:

  Aumente a concorrência da tarefa de sincronização em lote e planeje adequadamente o volume de dados para garantir que a tarefa seja concluída dentro de 24 horas.

• Mensagem de erro:

  Code:[OdpsWriter-09], Description:[Failed to write data to the ODPS destination table.]. - Failed to write block:0 to the ODPS destination table, uploadId=[202012081517026537dc0b0160354b]. Please contact the ODPS administrator for assistance. - java.io.IOException: Error writing request body to server。

• Causa possível:

  • Possível causa 1: Exceção de tipo de dado, ou seja, os dados de origem não estão em conformidade com as especificações de tipo de dado do ODPS. Por exemplo, gravar o valor 4.2223 em um tipo de dado decimal(18,10) no ODPS.

  • Possível causa 2: Exceção de bloco ODPS ou de comunicação.

• Solução:

  Converta os tipos de dados e utilize dados que estejam em conformidade com as especificações de tipo.

• Cenário: Uma tarefa de sincronização em lote de banco completo para MaxCompute falha com ErrorCode=MethodNotAllowed, ErrorMessage=cdc mode not supported.

• Causa possível: Os atributos transacionais ou de captura de dados alterados (CDC) da tabela MaxCompute de destino podem não corresponder ao modo de gravação utilizado pela tarefa de sincronização.

• Solução: Verifique se a tabela de destino foi criada com atributos CDC ou transacionais incompatíveis com o modo de gravação da tarefa de sincronização atual. Se não tiver certeza sobre os atributos da tabela, entre em contato com o suporte técnico para confirmação adicional.

• Cenário: Ao configurar uma tarefa de sincronização em lote com MaxCompute como destino, o seletor Tunnel resource group na configuração do destino não lista um Serverless resource group adquirido.

• Causa: O grupo de recursos Tunnel e o grupo de recursos que executa a tarefa de sincronização, como um Serverless resource group, são dois itens de configuração independentes com finalidades distintas. O grupo de recursos Tunnel é usado exclusivamente para transmissão de upload e download de dados do MaxCompute e utiliza, por padrão, o recurso público de transmissão, ou seja, a cota MaxCompute Tunnel. O grupo de recursos que executa a tarefa serve apenas para rodar a tarefa de sincronização em si, incluindo leitura da origem, processamento de dados e agendamento. Como as duas configurações atendem a propósitos diferentes, elas são independentes e não podem ser usadas de forma intercambiável.

Consulte o seguinte documento para obter detalhes sobre a configuração: Synchronize sharded MySQL tables to MaxCompute.

Adicione a fonte de dados usando uma string de conexão. Recomendamos modificar a URL JDBC para: jdbc:mysql://xxx.x.x.x:3306/database?com.mysql.jdbc.faultInjection.serverCharsetIndex=45. Para mais informações, consulte Add a MySQL data source.

• Causa do erro:

  • net_read_timeout: O DataX divide os dados do MySQL em várias instruções SELECT de tamanho igual com base no SplitPk. Durante a execução, uma das instruções SQL excede o tempo máximo de execução permitido no lado do RDS.

  • net_write_timeout: O tempo limite de espera para enviar um bloco ao cliente está configurado com um valor muito baixo.

• Solução:

  Adicione o parâmetro à URL de conexão da fonte de dados, defina net_write_timeout/net_read_timeout com um valor maior ou ajuste o parâmetro no console do RDS.

• Sugestão de melhoria:

  Se a tarefa puder ser reexecutada, configure-a para reexecução automática em caso de erro.

Por exemplo: jdbc:mysql://192.168.1.1:3306/lizi?useUnicode=true&characterEncoding=UTF8&net_write_timeout=72000

Causa do erro:

O parâmetro do MySQL wait_timeout tem como padrão 8 horas. Se os dados ainda estiverem sendo buscados quando esse tempo limite for atingido, a tarefa de sincronização será interrompida.

Solução:

Modifique o arquivo de configuração do MySQL my.cnf (ou my.ini no Windows). Adicione o parâmetro sob o módulo do MySQL (em segundos): wait_timeout=2592000 interactive_timeout=2592000. Em seguida, reinicie e faça login no MySQL, e execute a seguinte instrução para verificar: show variables like '%wait_time%'.

Uso normal de CPU, mas alto uso de memória, pode causar o encerramento da conexão.

Se você confirmar que a tarefa pode ser reexecutada automaticamente, recomendamos ativar a Reexecução Automática em Caso de Erro. Para mais informações, consulte Configure auto rerun.

• Cenário: Ao sincronizar dados do PostgreSQL com a ferramenta de sincronização em lote, ocorre o seguinte erro: org.postgresql.util.PSQLException: FATAL: terminating connection due to conflict with recovery

• Possível causa: Esse erro ocorre porque a extração de dados do banco de dados demora muito. Aumente os valores de max_standby_archive_delay e max_standby_streaming_delay. Para mais informações, consulte Standby Server Events.

• Cenário: Uma tarefa de sincronização em tempo real do AWS PostgreSQL para o MaxCompute falha porque o usuário do PostgreSQL de origem não possui o privilégio REPLICATION.

• Possível causa: O usuário do PostgreSQL de origem não tem o privilégio REPLICATION. Em algumas instâncias gerenciadas do AWS RDS PostgreSQL, esse atributo não pode ser concedido usando ALTER ROLE.

• Solução: A sincronização em lote (offline, agendada) não exige o privilégio REPLICATION. Se não for possível conceder o privilégio REPLICATION na sua instância do AWS PostgreSQL, utilize uma tarefa de sincronização em lote com agendamento periódico em vez da sincronização em tempo real como solução alternativa.

• Cenário: Após sincronizar uma coluna timestamp (sem fuso horário) do PostgreSQL para uma coluna DATETIME do MaxCompute, o valor resultante apresenta um desvio, por exemplo, de 2 horas, em relação ao valor esperado.

• Possível causa: O tipo timestamp (sem fuso horário) do PostgreSQL armazena o valor da hora local exatamente como recebido. Já o tipo DATETIME do MaxCompute armazena valores em UTC e os converte para exibição com base no fuso horário do projeto ou da sessão. Essa diferença no comportamento de armazenamento e conversão pode causar um desvio de fuso horário.

• Solução:

  1. Verifique se o fuso horário do servidor PostgreSQL, obtido executando SHOW timezone;, corresponde ao fuso horário do projeto MaxCompute. Você pode visualizar o fuso horário do projeto na página Basic Information do projeto MaxCompute.

  2. Se os fusos horários forem diferentes, configure o fuso horário nas configurações avançadas da tarefa de sincronização em lote. Defina o fuso horário da tarefa de sincronização para corresponder ao fuso horário do servidor PostgreSQL, de modo que os valores timestamp sejam analisados nesse fuso horário antes de serem gravados no MaxCompute, eliminando o desvio.

     Uma incompatibilidade de fuso horário entre a origem e o destino geralmente se manifesta como um deslocamento fixo de N horas em todos os campos de tempo. Por exemplo, ambas as extremidades estão definidas como Asia/Bangkok, mas os valores sincronizados diferem em 1 hora, ou o grupo de recursos executa na Alemanha e usa Europe/Berlin como padrão, enquanto seu negócio exige que os valores sejam armazenados em UTC. Nesses casos, selecione o fuso horário que você pretende usar nas configurações avançadas da tarefa.

  Nota

  Alterar o fuso horário de agendamento não afeta o fuso horário usado pelo processo do Data Integration. As duas configurações são independentes. Se as colunas de data ainda retornarem valores inesperados após ajustar o fuso horário de agendamento, defina o fuso horário explicitamente também nas configurações avançadas da tarefa de sincronização em lote.

  A configuração de fuso horário tem como padrão GMT+8. Mantenha o valor padrão quando os campos de tempo forem sincronizados corretamente e defina-o para o fuso horário esperado pelo seu negócio apenas quando ocorrer um deslocamento de tempo. No modo script, essa configuração equivale ao seguinte trecho:

  "common":{"column":{"timeZone":"Asia/Bangkok"}}

• Cenário: Ao configurar uma condição WHERE ou uma instrução querySql personalizada para sincronização incremental do PostgreSQL, é necessário converter um parâmetro de tempo em formato de string para timestamp para fins de comparação.

• Solução: Utilize a função padrão do PostgreSQL TO_TIMESTAMP em vez da função STR_TO_DATE do MySQL, pois o PostgreSQL e o MySQL usam dialetos SQL diferentes para converter strings de data e hora. Por exemplo:

  TO_TIMESTAMP('${start_time}', 'YYYYMMDDHH24')
  TO_TIMESTAMP('${end_time}', 'YYYYMMDDHH24')

  A função TO_TIMESTAMP converte uma string no formato especificado em um objeto timestamp, que pode então ser usado para filtrar linhas dentro de um intervalo de tempo na condição WHERE ou na instrução querySql.

• Cenário: Quando uma tarefa de sincronização em lote lê dados do Oracle com uma condição WHERE, a tarefa falha com ORA-00932: inconsistent datatypes.

• Possível causa: A cláusula WHERE compara diretamente uma coluna DATE do Oracle com um valor NUMBER, por exemplo, um literal de data escrito como um número simples. Essa comparação causa uma incompatibilidade de tipo de dados no Oracle.

• Solução: Converta explicitamente o valor numérico da data para o tipo DATE antes da comparação. Por exemplo, use TO_DATE('20250611', 'YYYYMMDD') ou o literal DATE '2025-06-11' na condição WHERE em vez de comparar a coluna com um número simples.

Ao conectar-se ao Amazon RDS e receber o erro Host is blocked, desative a verificação de integridade do balanceador de carga da Amazon. Após desativá-la, o problema de bloqueio não ocorrerá mais.

Ao adicionar uma fonte de dados MongoDB, utilize um usuário criado no banco de dados que contém as tabelas a serem sincronizadas. O usuário root não é suportado.

Por exemplo, se você deseja importar a tabela name e ela está no banco de dados test, o nome do banco de dados deve ser test, e você precisa usar o nome de um usuário criado no banco de dados test.

Use um nó de atribuição para primeiro converter um valor do tipo data em timestamp e, em seguida, passe esse valor como parâmetro de entrada para a tarefa de sincronização de dados do MongoDB.

Defina o fuso horário na configuração do MongoDB Reader. Para mais informações, consulte MongoDB Reader.

Reinicie a tarefa após um atraso sem alterar as condições de consulta, ou seja, adie o horário de execução da tarefa mantendo a configuração inalterada.

Durante a leitura de dados, o Column.name configurado pelo usuário diferencia maiúsculas de minúsculas. Uma configuração incorreta faz com que os dados lidos sejam nulos. Por exemplo:

• Dados de origem do MongoDB:

  {
      "MY_NAME": "zhangsan"
  }

• Configuração de colunas da tarefa de sincronização:

  {
      "column":
      [
          {
              "name": "my_name"
          }
      ]
  }

Como a grafia (maiúsculas/minúsculas) da configuração da coluna não corresponde aos dados de origem, a leitura dos dados falha.

O parâmetro de configuração de tempo limite é cursorTimeoutInMs, cujo padrão é 600000 ms (10 minutos). Esse parâmetro especifica o tempo total que o MongoDB Server gasta executando a consulta, excluindo o tempo de transferência de dados. Se a leitura completa dos dados for grande, o seguinte erro poderá ocorrer: MongoDBReader$Task - operation exceeded time limitcom.mongodb.MongoExecutionTimeoutException: operation exceeded time limit.

Atualmente, as tarefas de sincronização do DataWorks não suportam leitura de dados de um nó secundário. Se você configurar um nó secundário para leitura, ocorrerá o seguinte erro: no master.

• Análise da causa raiz:

  Causado por tempo limite do cursor.

• Solução:

  Aumente o valor do parâmetro cursorTimeoutInMs.

Aumente a simultaneidade da tarefa e o BatchSize de leitura.

• Possível causa:

  Por padrão, a tarefa de sincronização usa o comando splitVector para fragmentação da tarefa. Algumas versões do MongoDB não suportam o comando splitVector, o que causa o erro no such cmd splitVector.

• Solução:

  1. Acesse a página de configuração da tarefa de sincronização e clique no botão Converter para Script na parte superior. Altere a tarefa para o modo script.

  2. Na configuração de parâmetros do MongoDB, adicione o seguinte parâmetro:

     "useSplitVector" : false

     Isso evita o uso do splitVector.

• Mensagem de erro:

  Na tarefa de sincronização, tomando o modo assistente como exemplo, esse problema pode ocorrer quando o Write Mode (Overwrite) está definido como Yes e uma coluna que não é _id está configurada como Business Key.

  Na configuração de destino da tarefa de sincronização, selecione a fonte de dados MongoDB (nome da instância: xc_mongo_rds), defina o nome da coleção como xc_timestamp, ative o modo de sobrescrita (defina 'Overwrite' como Yes) e especifique my_id como chave primária de negócio."

• Possível causa:

  Os dados que estão sendo gravados contêm registros onde o _id não corresponde à Business Key configurada (como my_id no exemplo acima).

• Solução:

  • Opção 1: Modifique a tarefa de sincronização em lote para garantir que a Business Key configurada seja a mesma que _id.

  • Opção 2: Use _id como chave primária de negócio durante a sincronização de dados.

• Causa:

  Quando o Redis usa o modo hash para armazenamento, os atributos e valores do hash devem aparecer em pares. Por exemplo: odpsReader: "column":[ "id", "name", "age", "address" ]. No destino, se o RedisWriter estiver configurado como: "keyIndexes":[ 0, 1], então no Redis, id e name servem como chave, age serve como atributo e address serve como valor no tipo hash. Se apenas duas colunas forem configuradas na origem ODPS, o modo hash não poderá ser usado para armazenamento no Redis, e essa exceção será lançada.

• Solução:

  Se quiser usar apenas duas colunas, configure o modo String do Redis para armazenamento. Caso precise usar o modo hash, configure pelo menos três colunas no lado da origem.

• Sintoma:

  Ao configurar uma tarefa de sincronização em lote para ler dados de armazenamento de arquivos como OSS ou FTP, se o arquivo estiver no formato CSV e usar vários caracteres como delimitador de coluna (como |,, ## ou ;;), a tarefa poderá falhar com um erro de dados sujos. No log de execução, você verá um erro IndexOutOfBoundsException juntamente com dados sujos.

• Análise da causa raiz:

  O leitor csv interno ("fileFormat": "csv") no DataWorks tem limitações ao processar delimitadores de múltiplos caracteres, o que causa uma divisão imprecisa de colunas nas linhas de dados.

• Solução:

  • Modo assistente: Altere o tipo de texto para text e especifique explicitamente o delimitador de múltiplos caracteres.

  • Modo script: Altere "fileFormat": "csv" para "fileFormat": "text" e defina corretamente o delimitador: "fieldDelimiter":"<multi-char delimiter>", "fieldDelimiterOrigin":"<multi-char delimiter>".

A sincronização em lote em si não limita o número de arquivos lidos pelo plugin OSS Reader. A principal limitação vem dos recursos de CU consumidos pela tarefa. Ler muitos arquivos de uma só vez pode facilmente causar erros de falta de memória. Portanto, não recomendamos configurar o parâmetro object como: *, para evitar erros OutOfMemoryError: Java heap space .

O OSS Writer grava nomes de arquivos simulando diretórios por meio de nomes de objetos. O OSS possui restrições quanto aos nomes dos objetos. Ao usar "object": "datax", os objetos gravados começam com datax, com sufixos de strings aleatórias anexados. O número de arquivos é determinado pelo número real de tarefas divididas.

Se não precisar de sufixos UUID aleatórios, configure "writeSingleObject" : "true". Para mais informações, consulte a descrição do parâmetro writeSingleObject na documentação OSS Writer.

• Causa:

  A AccessKey configurada para a fonte de dados não tem permissões no bucket.

• Solução:

  Conceda permissões de leitura no bucket à conta AccessKey configurada para a fonte de dados OSS.

• Análise da causa raiz:

  O parâmetro mapred.task.timeout pode estar definido com um valor muito baixo, fazendo com que o Hadoop encerre a tarefa e limpe o diretório temporário, tornando os dados temporários indisponíveis.

• Solução:

  Na seção de fonte de dados da tarefa de sincronização em lote, se Hive read methods estiver definido como Read Data Based on Hive JDBC (Supports Conditional Filtering), defina o valor do parâmetro mapred.task.timeout em Session Configuration, por exemplo, mapred.task.timeout=600000.

• Mensagem de erro:

  ERROR JobContainer - Exception when job runcom.alibaba.datax.common.exception.DataXException: Code:[DatahubWriter-04], Description:[Failed to write data.]. - com.aliyun.datahub.exception.DatahubServiceException: Record count 12498 exceed max limit 10000 (Status Code: 413; Error Code: TooLargePayload; Request ID: 20201201004200a945df0bf8e11a42)

• Possível causa:

  O erro ocorre porque o volume de dados enviado pelo DataX ao DataHub em um único lote excede o limite do DataHub. Os principais parâmetros de configuração que afetam o volume de dados enviado ao DataHub são:

  • maxCommitSize: Especifica o tamanho acumulado dos dados no buffer. Quando os dados acumulados atingem o maxCommitSize (em MB), eles são enviados ao destino em um lote. O padrão é 1 MB (1.048.576 bytes).

  • batchSize: Especifica a contagem acumulada de registros de dados no buffer para o DataX-On-Flume. Quando a contagem acumulada de registros atinge o batchSize, os dados são enviados ao destino em um lote.

• Solução:

  Reduza os valores dos parâmetros maxCommitSize e batchSize.

Este plugin diferencia maiúsculas de minúsculas nos nomes das colunas. Verifique a configuração de colunas do LogHub Reader.

O Data Integration utiliza o momento em que os dados entram no LogHub. Verifique no console do LogHub se a coluna de metadados receive_time está dentro do intervalo de tempo configurado para a tarefa.

Se isso ocorrer, edite manualmente a configuração de colunas na interface.

A hora inicial e final configuradas na tarefa de sincronização em lote são usadas pelo Reader para chamar a API GetCursor do SLS e localizar os cursores inicial e final. Esse tempo é usado para localizar o intervalo de leitura com base no tempo de recebimento do lado do servidor SLS. A tarefa realmente lê dados dentro do intervalo do cursor, o que não equivale a filtrar pela coluna de saída __time__.

A coluna de saída __time__ vem de log.getTime() de cada entrada de log, representando o próprio tempo do log. As consultas no console do SLS normalmente usam o intervalo de tempo da consulta, instruções de consulta e colunas de índice para estatísticas, geralmente baseadas no tempo do log __time__. Portanto, mesmo que a tarefa de sincronização e o console usem os mesmos valores de tempo, o intervalo __time__ ou a contagem de registros pode diferir se os dois lados usarem métricas de tempo diferentes.

Cenários comuns:

1. Quando há atraso na coleta ou entrega de logs, preenchimento retroativo de logs históricos ou relógios de cliente imprecisos, o tempo do log __time__ pode ser anterior ou posterior ao tempo de recebimento do lado do servidor SLS. A tarefa de sincronização localiza cursores com base no tempo de recebimento do servidor, enquanto o console faz consultas com base em __time__, o que pode gerar resultados diferentes.

2. Quando dados são gravados em outro LogStore por meio de transformação de dados do SLS, se a instrução de transformação não definir explicitamente __time__, o __time__ do log de destino geralmente retém o tempo do log de origem em vez do tempo de execução da transformação. Nesse caso, a tarefa de sincronização pode ler esse lote de dados dentro do intervalo de tempo em que a transformação grava no LogStore de destino. No entanto, ao consultar o console do LogStore de destino pelo tempo de execução da transformação ou pelo intervalo de tempo atual, esses logs podem não ser encontrados. É necessário consultar pelo intervalo __time__ real dos logs.

3. Quando a instrução de consulta do console, as colunas de índice, o intervalo de tempo e a instrução de filtragem de regras (SPL) na tarefa de sincronização são inconsistentes, as contagens de registros podem diferir mesmo que as métricas de tempo sejam as mesmas.

Sugestões de solução de problemas:

1. Verifique se o intervalo de tempo da consulta no console, a instrução de consulta, as colunas de índice e a hora inicial/final e a instrução de filtragem de regras (SPL) na tarefa de sincronização estão consistentes.

2. Inclua tanto __time__ (tempo do log) quanto __tag__:__receive_time__ (o campo observável para o tempo de recebimento do lado do servidor SLS, que requer que este campo exista nas tags do log) na configuração column para comparar o tempo do log com o tempo de recebimento do servidor.

3. Se os dados vierem de transformação de dados do SLS, verifique se a instrução de transformação define explicitamente __time__ e ajuste o intervalo de tempo da consulta no console no LogStore de destino com base no __time__ real.

4. Se for necessária uma reconciliação estrita pelo tempo do log no downstream, filtre ou agregue por __time__ após a gravação no destino.

Exemplo: O __time__ do log de origem é 2026-06-01 10:00:00. Uma tarefa de transformação de dados do SLS grava este log no LogStore de destino às 2026-06-12 10:00:00 sem modificar explicitamente o __time__. O __time__ do log de destino permanece 2026-06-01 10:00:00. Se as horas inicial e final da tarefa de sincronização cobrirem 2026-06-12 10:00:00, a tarefa poderá ler este log. No entanto, ao consultar o LogStore de destino no console por volta de 2026-06-12 10:00:00 com __time__ como filtro, este log pode não ser encontrado. Nesse caso, ajuste o tempo da consulta no console para cerca de 2026-06-01 10:00:00, ou defina explicitamente o __time__ do log de destino durante a transformação de dados conforme necessário.

O Reader corresponde aos nomes das colunas a partir dos campos reais do conteúdo do log extraído, mapeamentos de metacampos internos do Reader e LogTag com base na configuração column. Os nomes das colunas diferenciam maiúsculas de minúsculas. Se nenhuma correspondência for encontrada, null é gerado sem erro.

As causas comuns incluem:

1. O nome da coluna configurado em column tem grafia (maiúsculas/minúsculas) diferente da chave original do campo de log.

2. O console exibe aliases de análise de consulta, campos de índice ou campos expandidos de JSON, que diferem da chave de log original que o Reader realmente recupera.

3. A coluna realmente vem do LogTag e precisa ser configurada como __tag__:<tagKey>.

4. Após configurar uma instrução de filtragem de regras (SPL) ou transformação, os nomes dos campos de saída não correspondem totalmente à configuração column.

Ao solucionar problemas, verifique primeiro as colunas da tabela de origem e a visualização de dados na página visual para confirmar os campos que o Reader realmente identifica. No modo script, você também pode definir temporariamente column como ["*"] para ver as chaves reais dos campos de conteúdo do log recuperadas pelo Reader e, em seguida, configurar column com base nas chaves originais.

O comportamento é o mesmo da lógica de escrita da API: dados na mesma linha e mesma coluna são sobrescritos, e outros dados permanecem inalterados.

Recupere o mapeamento do índice ES usando o comando curl e extraia todas as colunas do mapeamento.

• Comando Shell para consulta:

  //es7
  curl -u username:password --request GET 'http://esxxx.elasticsearch.aliyuncs.com:9200/indexname/_mapping'
  //es6
  curl -u username:password --request GET 'http://esxxx.elasticsearch.aliyuncs.com:9200/indexname/typename/_mapping'

• Recuperando colunas do resultado:

  {
      "indexname": {
          "mappings": {
              "typename": {
                  "properties": {
                      "field1": {
                          "type": "text"
                      },
                      "field2": {
                          "type": "long"
                      },
                      "field3": {
                          "type": "double"
                      }
                  }
              }
          }
      }
  }

  As colunas e definições de atributos sob properties na resposta representam todas as colunas do índice. Por exemplo, o índice acima contém três colunas: field1, field2 e field3.

Adicione parâmetros de agendamento de data à configuração do índice para calcular automaticamente a string do índice com base em datas diferentes, permitindo alterações automáticas no nome do índice do Elasticsearch Reader. A configuração envolve três etapas: definir parâmetros de data, configurar parâmetros de índice e implantar e executar a tarefa.

1. Defina parâmetros de data: Nas configurações de agendamento da tarefa de sincronização, adicione parâmetros para definir os parâmetros de data. A seguinte configuração var1 representa o tempo de execução da tarefa (dia atual), e var2 representa a data de negócio (dia anterior).

2. Configure parâmetros de índice: Mude a tarefa para o modo script e configure o índice do Elasticsearch Reader usando o formato: ${variable_name}, conforme mostrado abaixo.

   {
       "type": "job",
       "version": "2.0",
       "steps": [
           {
               "stepType": "elasticsearch",
               "parameter": {
                   "retryCount": 30,
                   "scroll": "10m",
                   "column": [
                       "col18",
                       "col17"
   
                   ],
                   "index": "esstress_1_${var1}_${var2}",
                   "pageSize": 100,
                   "sort": {
                       "_id": "asc"
                   },

3. Implante e execute a tarefa: Após a verificação, envie e implante a tarefa no Operation Center e execute-a como um agendamento periódico ou tarefa de backfill de dados.

   1. Clique no botão Running with Parameters para executar a tarefa diretamente para verificação. A execução com parâmetros substitui os parâmetros do sistema de agendamento usados na configuração da tarefa. Após a execução, verifique os logs para confirmar se o índice sincronizado atende às expectativas.

      Nota

      Ao executar com parâmetros, insira os valores dos parâmetros diretamente para teste de substituição.

   2. Se a etapa anterior for verificada conforme o esperado, a configuração da tarefa estará completa. Clique em Save e depois em Commit para enviar a tarefa de sincronização para o ambiente de produção.

      Em um workspace no modo padrão, clique em Deploy para acessar o Deployment Center e implantar a tarefa de sincronização no ambiente de produção.

4. Resultado: A seguir, são apresentados a configuração e o resultado real do índice em tempo de execução.

   Configuração do índice no script: "index": "esstress_1_${var1}_${var2}".

   Índice resolvido em tempo de execução: esstress_1_20230106_20230105.

   ],
   "full":false,
   "gmtCreate":"2022-07-18 14:47:18",
   "gmtModified":"2022-07-18 14:47:18",
   "index":"esstress_1_20230106_20230105",
   "instanceId":"es-cn-2r42se1je001zwmt0",
   "ownerId":"1224800975333052",
   "pageSize":100,
   "password":"********",
   "privateNetworkIpWhiteList":[
       "0.0.0.0/0"
   ],

Para sincronizar propriedades de campos do tipo objeto, utilize exclusivamente o modo script. Nesse modo, configure multi conforme abaixo e especifique column usando o formato atributo.subatributo.

"multi":{
   "multi":true 
 }

Consulte o exemplo a seguir para a configuração:

#Example:
##Data in Elasticsearch
"hits": [
    {
        "_index": "mutiltest_1",
        "_type": "_doc",
        "_id": "7XAOOoMB4GR_1Dmrrust",
        "_score": 1.0,
        "_source": {
            "level1": {
                "level2": [
                    {
                        "level3": "testlevel3_1"
                    },
                    {
                        "level3": "testlevel3_2"
                    }
                ]
            }
        }
    }
]
##Reader configuration
"parameter": {
  "column": [
      "level1",
      "level1.level2",
      "level1.level2[0]"
  ],
  "multi":{
        "multi":true
    }
}
##Writer result: 1 row with 3 columns, column order matches reader configuration
COLUMN              VALUE
level1:             {"level2":[{"level3":"testlevel3_1"},{"level3":"testlevel3_2"}]}
level1.level2:      [{"level3":"testlevel3_1"},{"level3":"testlevel3_2"}]
level1.level2[0]:   {"level3":"testlevel3_1"}

1. As aspas duplas extras exibidas antes e depois dos caracteres são um problema de visualização no Kibana. Os dados reais não possuem essas aspas duplas no início e no fim. Utilize o comando curl ou o Postman para visualizar os dados reais. O comando curl para recuperar os dados é o seguinte:

   //es7
   curl -u username:password --request GET 'http://esxxx.elasticsearch.aliyuncs.com:9200/indexname/_mapping'
   //es6
   curl -u username:password --request GET 'http://esxxx.elasticsearch.aliyuncs.com:9200/indexname/typename/_mapping'

2. Configure o tipo da coluna de escrita no ES como nested para sincronizar dados de string do tipo JSON do ODPS para o ES no formato aninhado. O exemplo a seguir sincroniza a coluna name para o ES no formato nested.

   • Configuração de sincronização: Defina o tipo de name como nested.

   • Resultado da sincronização: name é um tipo de objeto aninhado.

     "total": {
         "value": 1,
         "relation": "eq"
     },
     "max_score": 1.0,
     "hits": [
         {
             "_index": "test",
             "_type": "_doc",
             "_id": "bb5oqoUBlqHPyI16REEQ",
             "_score": 1.0,
             "_source": {
                 "name": [
                     {
                         "fields1": "value"
                     }
                 ]
             }
         }
     ]
     }

Existem dois métodos de configuração para gravar tipos de array no ES. Escolha o método de sincronização correspondente com base no formato dos dados da source.

• Grave no ES como um tipo de array analisando os dados da source como JSON. Por exemplo, se os dados da source forem "[1,2,3,4,5]", configure json_array=true para analisar os dados da source e gravá-los na coluna do ES como um array. Configure o ColumnList com json_array=true.

  • Configuração no modo assistente.

  • Configuração no modo script:

    "column":[
      {
        "name":"docs",
        "type":"keyword",
        "json_array":true
      }
    ]

• Grave no ES como um tipo de array analisando os dados da source com um delimitador. Por exemplo, se os dados da source forem "1,2,3,4,5", configure um delimitador splitter="," para analisar e gravar os dados na coluna do ES como um array.

  • Limitações:

    • Uma tarefa suporta apenas um delimitador. O splitter é globalmente único e não aceita delimitadores diferentes para colunas de array distintas. Por exemplo, para as colunas da source col1="1,2,3,4,5" , col2="6-7-8-9-10", não é possível configurar o splitter separadamente para cada coluna.

    • O splitter pode ser configurado como uma expressão regular. Por exemplo, se o valor da coluna da source for "6-,-7-,-8+,*9-,-10", você pode configurar splitter:".,.", sendo isso suportado também no modo assistente.

  • Configuração no modo assistente: splitter: o padrão é "-,-".

  • Configuração no modo script:

    "parameter" : {
          "column": [
            {
              "name": "col1",
              "array": true,
              "type": "long"
            }
          ],
          "splitter":","
    }

• Análise da causa raiz:

  O HttpClient determina que, sempre que uma conexão é estabelecida, uma solicitação não autenticada é enviada primeiro. Após o servidor retornar a exigência de autenticação (especificando o método com base na resposta), uma solicitação autenticada é então realizada. Como cada gravação de dados no ES exige o estabelecimento de uma conexão, cada operação gera uma solicitação não autenticada, que fica registrada nos logs de auditoria.

• Solução:

  Adicione a configuração "preemptiveAuth":true no modo script.

Existem dois métodos para configurar a gravação de datas. Escolha o mais adequado conforme sua necessidade.

• Grave diretamente na coluna Date do ES com base no conteúdo lido pelo Reader:

  • Configure origin:true para gravar o conteúdo lido diretamente no ES.

  • Configure "format" para especificar o atributo de formato da coluna ao criar o mapping por meio da gravação no ES.

    "parameter" : {
        "column": [
            {
                "name": "col_date",
                "type": "date",
                "format": "yyyy-MM-dd HH:mm:ss",
                "origin": true
            }
              ]
    }

• Conversão de fuso horário: Se for necessário que o Data Integration realize a conversão de fuso horário, adicione o parâmetro Timezone.

  "parameter" : {
      "column": [
          {
              "name": "col_date",
              "type": "date",
              "format": "yyyy-MM-dd HH:mm:ss",
              "Timezone": "UTC"
          }
            ]
  }

• O type:version está configurado, mas o ES não suporta a especificação de uma versão externa.

      "column":[
                              {
                                  "name":"id",
                                  "type":"version"
                              },
    ]

• Solução:

  Remova a configuração "type":"version". O Elasticsearch Writer não suporta a especificação de versão externa.

• Análise da causa raiz:

  O Elasticsearch Reader não consegue analisar o formato de data de uma coluna do tipo date porque o mapping da coluna de data correspondente no ES não tem um formato configurado.

• Solução:

  • Configure o parâmetro dateFormat com o mesmo formato da coluna de data do ES, usando "||" como separador. O formato deve incluir todos os formatos de tipo de data. Por exemplo:

    "parameter" : {
          "column": [
         			"dateCol1",
            	"dateCol2",
              "otherCol"
          ],
         "dateFormat" : "yyyy-MM-dd||yyyy-MM-dd HH:mm:ss",
    }

  • Defina o formato de mapping para todas as colunas de data no banco de dados do ES.

• Análise da causa raiz:

  Devido às limitações de palavras-chave do fastjson, o índice ou as colunas podem conter palavras-chave como $ref.

• Solução:

  O Elasticsearch Reader não suporta a sincronização de índices que contenham a palavra-chave $ref nos nomes das colunas. Para mais informações, consulte Elasticsearch Reader.

• Análise da causa raiz:

  Isso acionou o mecanismo de bloqueio otimista do ES. O número da versão atual deveria ter um valor específico, mas o número da versão passado pelo comando update é diferente, causando um conflito de versão. Durante a atualização, outra operação estava excluindo dados do índice simultaneamente.

• Solução:

  1. Verifique se há operações de exclusão de dados em andamento.

  2. Altere o método de sincronização da tarefa de Update para Index.

• Análise da causa raiz:

  Ao configurar atributos avançados como similarity e properties para uma coluna, other_params é necessário para que o plugin os reconheça.

  "parameter":{
          "__datasource__type":"elasticsearch",
          "actionType":"index",
          "aliasMode":"append",
          "batchSize":1024,
          "cleanup":false,
          "column":[
              {
                  "name":"id",
                  "type":"long"
              },
              {
                  "name":"dim1_code",
                  "type":"keyword"
              },
              {
                  "analyzer":"china",
                  "name":"dim1_name",
                  "similarity":"len_similarity",
                  "type":"text"
              },
              {
                  "analyzer":"china",
                  "name":"dim1_val",
                  "similarity":"len_similarity",
                  "type":"text"
              },
              {
                  "name":"dim1_sort",
                  "type":"long"
              }

• Solução:

  Configure other_params na configuração da coluna e adicione similarity dentro de other_params, conforme abaixo:

  {"name":"dim2_name",...,"other_params":{"similarity":"len_similarity"}}

• Análise da causa raiz:

  Atualmente, a gravação de sincronização em lote no Elasticsearch não suporta o tipo dense_vector. Apenas os seguintes tipos são suportados:

  ID,PARENT,ROUTING,VERSION,STRING,TEXT,KEYWORD,LONG,
  INTEGER,SHORT,BYTE,DOUBLE,FLOAT,DATE,BOOLEAN,BINARY,
  INTEGER_RANGE,FLOAT_RANGE,LONG_RANGE,DOUBLE_RANGE,DATE_RANGE,
  GEO_POINT,GEO_SHAPE,IP,IP_RANGE,COMPLETION,TOKEN_COUNT,OBJECT,NESTED;

• Solução:

  Para tipos não suportados pelo Elasticsearch Writer, proceda da seguinte forma:

  • Não recomendamos usar o Elasticsearch Writer para criar mappings de índices. Utilize um mapping personalizado.

  • Altere o tipo correspondente para NESTED.

  • Modifique a configuração para: dynamic = true, cleanup=false.

• Causa:

  #Incorrect configuration
  "settings": {
    "index": {
      "number_of_shards": 1,
      "number_of_replicas": 0
    }
  }
  #Correct configuration
  "settings": {
    "number_of_shards": 1,
    "number_of_replicas": 0
  }

• Solução:

  As configurações de Settings entram em vigor apenas quando um índice é criado, o que abrange dois casos: o índice não existe ou cleanup=true. Quando cleanup=true, a configuração de Settings não precisa incluir "index".

#Original mappings
{
  "name":"box_label_ret",
  "properties":{
    "box_id":{
      "type":"keyword"
    }
}
#After rebuilding with cleanup=true, it becomes
{
    "box_label_ret": {
      "properties": {
        "box_id": {
          "type": "text",
          "fields": {
            "keyword": {
              "type": "keyword",
              "ignore_above": 256
            }}}}
}

• Análise da causa raiz:

  Para tipos aninhados, o Elasticsearch Writer utiliza apenas os mappings de nível superior e permite que o ES adapte automaticamente os tipos complexos aninhados. A alteração do tipo de atributo para text com a adição de fields:keyword é um comportamento de adaptação automática do ES e não afeta seu uso. Caso necessite de um formato de mapping específico, consulte Elasticsearch Writer.

• Solução:

  Crie os mappings de índice esperados no ES antes da sincronização, defina cleanup como false na tarefa de sincronização do ES e execute a tarefa.

O Kafka Reader lê dados em lotes. Em um lote de dados, se algum registro exceder o endDateTime, a sincronização é interrompida. No entanto, os dados que ultrapassam o endDateTime nesse lote ainda são gravados na source de dados de destino.

• Use a configuração skipExceedRecord para especificar se os dados excedentes devem ser sincronizados. Para detalhes de uso, consulte Kafka Reader. [Não recomendamos definir para não sincronizar, pois isso pode causar perda de dados.]

• Configure o parâmetro max.poll.records do Kafka para especificar a quantidade de dados extraídos em cada lote. Combinado com a concorrência, você controla o volume de dados que pode exceder o limite. O volume de dados excedente < max.poll.records × concorrência.

• Análise da causa raiz:

  Quando o volume de dados é pequeno ou a distribuição é desigual, algumas partições do Kafka podem não receber novos dados ou os novos dados podem não atingir o offset final especificado. Como a condição de saída da tarefa exige que todas as partições atinjam o offset final especificado, essas partições "ociosas" não satisfazem a condição, impedindo a conclusão normal da tarefa.

• Solução:

  Defina a política de fim de sincronização como 1 minuto sem ler novos dados (no modo script, defina stopWhenPollEmpty como true e stopWhenReachEndOffset como true). A tarefa é encerrada após ler os dados do offset mais recente de todas as partições, evitando execução ociosa. Contudo, registros com timestamps anteriores ao offset final configurado que forem gravados após o término da tarefa não serão consumidos.

O RestAPI Writer oferece dois modos de gravação. Ao sincronizar vários registros, defina dataMode como multiData e adicione o parâmetro dataPath:"data.list" no script. Para mais informações, consulte RestAPI Writer.

1. A configuração do OTS Writer deve atender aos dois requisitos a seguir:

   "newVersion": "true",
   "enableAutoIncrement": "true",

2. O nome da coluna de chave primária com incremento automático não deve ser configurado no OTS Writer.

3. A soma das entradas de primaryKey e das entradas de column configuradas no OTS Writer deve ser igual ao número de colunas nos dados do OTS Reader upstream.

Exemplo: Um registro de dados possui três tags: [phone=xiaomi, RAM=8G, camera=LEICA].

• Exemplo de exportação de dados (OTS Reader)

  • Para mesclar as tags acima em uma única coluna para exportação, configure da seguinte forma:

    "column": [
          {
            "name": "_tags",
          }
        ],

    O DataWorks exporta as tags como uma única coluna de dados no seguinte formato:

    ["phone=xiaomi","camera=LEICA","RAM=8G"]

  • Para exportar a tag phone e a tag camera como colunas separadas, configure da seguinte forma:

    "column": [
          {
            "name": "phone",
            "is_timeseries_tag":"true",
          },
          {
            "name": "camera",
            "is_timeseries_tag":"true",
          }
        ],

    O DataWorks exporta duas colunas de dados no seguinte formato:

    xiaomi, LEICA

• Exemplo de importação de dados (OTS Writer)

  A source de dados upstream (Reader) possui duas colunas de dados:

  • Uma coluna contém: ["phone=xiaomi","camera=LEICA","RAM=8G"].

  • A outra coluna contém: 6499.

  Para adicionar ambas as colunas às tags, o formato esperado do campo de tag após a gravação é o seguinte: Configure da seguinte forma:

  "column": [
        {
          "name": "_tags",
        },
        {
          "name": "price",
          "is_timeseries_tag":"true",
        },
      ],

  • A configuração da primeira coluna importa ["phone=xiaomi","camera=LEICA","RAM=8G"] integralmente para o campo de tag.

  • A configuração da segunda coluna importa price=6499 individualmente para o campo de tag.

Se os nomes das suas tabelas seguirem um padrão regular, como orders_20170310, orders_20170311 e orders_20170312, onde as tabelas são diferenciadas por data e compartilham a mesma estrutura, use parâmetros de agendamento (Configure synchronization tasks in script mode) para personalizar o nome da tabela e ler automaticamente os dados da tabela do dia anterior da base de dados da source toda madrugada.

No modo script, altere o nome da tabela da source para uma variável, como orders_${tablename}. Como as tabelas são diferenciadas por data e você precisa ler os dados do dia anterior diariamente, atribua o valor da variável na configuração de parâmetros da tarefa como tablename=${yyyymmdd}.

Acesse a página de configuração da tarefa de sincronização, modifique os mapeamentos de colunas para atualizar as colunas alteradas na configuração da tarefa e, em seguida, reenvie e execute a tarefa para que as alterações entrem em vigor.

Ao configurar um nó de sincronização em lote, a seção Source exibe apenas as primeiras 25 tabelas da source de dados selecionada por padrão. Se houver mais tabelas, insira o nome da tabela para pesquisar ou utilize o modo script para desenvolvimento.

• Causa do erro: A configuração da coluna contém palavras-chave reservadas ou colunas que começam com um número.

• Solução: Mude a tarefa de sincronização do Data Integration para o modo script e faça o escape das colunas especiais na configuração de colunas. Para configurar tarefas no modo script, consulte Configure synchronization tasks in script mode.

  • O caractere de escape para MySQL é keyword.

  • O caractere de escape para Oracle e PostgreSQL é "keyword".

  • O caractere de escape para SQL Server é [keyword].

  Exemplo para MySQL:

  {
      "stepType": "mysql",
      "parameter": {
          "envType": 0,
          "datasource": "wpw_test_mysql",
          "column": [
              "id",
              "`order`",
              "`add`"
          ],
          "connection": [
              {
                  "datasource": "wpw_test_mysql",
                  "table": [
                      "abc"
                  ]
              }

• Tomando uma source de dados MySQL como exemplo:

  1. Execute a seguinte instrução para criar uma tabela chamada aliyun: create table aliyun (

  2. Execute a seguinte instrução para criar uma view e atribuir um alias à coluna da tabela: create view v_aliyun as select

     Nota

     • table é uma palavra-chave do MySQL. Durante a sincronização de dados, o código concatenado causa um erro. Crie uma view e atribua um alias à coluna da tabela.

     • Não recomendamos o uso de palavras-chave como nomes de colunas de tabela.

  3. Após executar as instruções acima, utilize a view v_aliyun em vez da tabela aliyun ao configurar a tarefa de sincronização.

Esse erro pode ocorrer porque o mapeamento de colunas da tarefa de sincronização não está configurado corretamente ou o plugin não tem column devidamente configurado.

1. Verifique se o mapeamento de colunas está configurado.

2. Verifique se o plugin tem column configurado corretamente.

• Sintoma:

  Ao clicar em Preview Data, uma mensagem semelhante à seguinte aparece, indicando que o tamanho em bytes da coluna excede o limite.

  A mensagem de erro é "Maximum column length of 1,000 exceeded in column 6 in record 1. Set the SafetySwitch property to false if you're expecting column lengths greater than 100,000 characters to avoid this error."

• Causa: Para evitar OOM, o service da source de dados verifica o comprimento da coluna ao processar solicitações de pré-visualização de dados. Se uma única coluna exceder 1000 bytes, a mensagem acima aparece. Essa mensagem não afeta a execução real da tarefa. Ignore este erro e execute a tarefa de sincronização em lote diretamente.

  Nota

  Se o arquivo existir e a conectividade estiver normal, as seguintes situações também podem causar falha na pré-visualização de dados:

  • Uma única linha no arquivo excede o limite de tamanho em bytes de 10 MB. Nesse caso, nenhum dado é exibido, de forma semelhante à mensagem acima.

  • Uma única linha no arquivo excede o limite de contagem de 1000 colunas. Nesse caso, apenas as primeiras 1000 colunas são exibidas, com uma mensagem mostrada na 1001ª coluna.

O TTL é definido no nível da tabela. Não há opção de TTL na configuração da tarefa de sincronização.

A sincronização baseada em API não suporta o uso de funções do lado da source. Processe os dados usando funções da source antes de importar.