Solução de problemas do conector Kafka¶

Bytes brutos¶

O nome de arquivo <nome_conector>/<nome_tabela>/<partição>/offset_(<chave>/<valor>_)<carimbo_data_hora>.gz inclui o offset exato do registro que não foi convertido de bytes brutos para o formato do arquivo de origem. Para resolver os problemas, reenvie o registro para o conector Kafka como um novo registro.

Formato do arquivo de origem (Avro, JSON ou Protobuf)¶

Se o Snowpipe não tiver conseguido carregar os dados dos arquivos no estágio interno criado para o tópico Kafka, o conector Kafka move os arquivos para o estágio da tabela de destino no formato do arquivo de origem.

Se um conjunto de arquivos tiver vários problemas, a coluna FIRST_ERROR_MESSAGE na saída COPY_HISTORY indica apenas o primeiro erro encontrado. Para visualizar todos os erros nos arquivos, é necessário recuperar os arquivos do estágio de tabela, carregá-los para um estágio nomeado, e depois executar uma instrução COPY INTO <tabela> com a opção de cópia VALIDATION_MODE definida como RETURN_ALL_ERRORS. A opção de cópia VALIDATION_MODE instrui uma instrução COPY para validar os dados a serem carregados e retornar resultados com base na opção de validação especificada. Nenhum dado é carregado quando essa opção de cópia é especificada. Na instrução, referencie o conjunto de arquivos que você tentou carregar usando o conector Kafka.

Quando o problema com os arquivos de dados for resolvido, você pode carregar os dados manualmente usando uma ou mais instruções COPY.

Os exemplos a seguir referenciam arquivos de dados localizados no estágio de tabela da tabela mytable no banco de dados e esquema mydb.public.

Para validar arquivos de dados no estágio de tabela e resolver erros:

1. Liste todos os arquivos localizados no estágio da tabela usando LIST.

   Por exemplo:

   LIST @mydb.public.%mytable;
   

   Copy

   Os exemplos nesta seção presumem que JSON é o formato dos arquivos de dados de origem.

2. Faça o download dos arquivos criados pelo conector Kafka na sua máquina local usando GET.

   Por exemplo, baixe os arquivos para um diretório chamado data na sua máquina local:

   Linux ou macOS:

   GET @mydb.public.%mytable file:///data/;
   

   Copy

   Microsoft Windows:

   GET @mydb.public.%mytable file://C:\data\;
   

   Copy

3. Usando CREATE STAGE, crie um estágio interno nomeado que armazene arquivos de dados com o mesmo formato do arquivos Kafka de origem.

   Por exemplo, crie um estágio interno chamado kafka_json que armazene arquivos JSON:

   CREATE STAGE kafka_json FILE_FORMAT = (TYPE = JSON);
   

   Copy

4. Carregue os arquivos que você baixou do estágio de tabela usando PUT.

   Por exemplo, carregue os arquivos baixados para o diretório data em sua máquina local:

   Linux ou macOS:

   PUT file:///data/ @mydb.public.kafka_json;
   

   Copy

   Microsoft Windows:

   PUT file://C:\data\ @mydb.public.kafka_json;
   

   Copy

5. Crie uma tabela temporária com duas colunas de variantes para fins de teste. A tabela é usada apenas para validar o arquivo de dados preparado. Nenhum dado é carregado na tabela. A tabela é descartada automaticamente quando a sessão atual do usuário termina:

   CREATE TEMPORARY TABLE t1 (col1 variant);
   

   Copy

6. Recupere todos os erros encontrados no arquivo de dados executando uma instrução COPY INTO *tabela* … VALIDATION_MODE = “RETURN_ALL_ERRORS”. A instrução valida o arquivo no estágio especificado. Nenhum dado é carregado na tabela:

   COPY INTO mydb.public.t1
     FROM @mydb.public.kafka_json
     FILE_FORMAT = (TYPE = JSON)
     VALIDATION_MODE = 'RETURN_ALL_ERRORS';
   

   Copy

7. Corrija todos os erros relatados nos arquivos de dados na sua máquina local.

8. Carregue os arquivos fixos para o estágio de tabela ou para o estágio interno nomeado usando PUT.

   O seguinte exemplo carrega os arquivos para o estágio de tabela, substituindo os arquivos existentes:

   Linux ou macOS:

   PUT file:///tmp/myfile.csv @mydb.public.%mytable OVERWRITE = TRUE;
   

   Copy

   Windows:

   PUT file://C:\temp\myfile.csv @mydb.public.%mytable OVERWRITE = TRUE;
   

   Copy

9. Carregue os dados na tabela de destino usando COPY INTO tabela sem a opção VALIDATION_MODE.

   Como alternativa, você pode usar a opção de cópia PURGE = TRUE para excluir os arquivos de dados do estágio quando os dados tiverem sido carregados com sucesso, ou apagar manualmente os arquivos do estágio de tabela usando REMOVE:

   COPY INTO mydb.public.mytable(RECORD_METADATA, RECORD_CONTENT)
     FROM (SELECT $1:meta, $1:content FROM @mydb.public.%mytable)
     FILE_FORMAT = (TYPE = 'JSON')
     PURGE = TRUE;
   

   Copy