Kafka 커넥터 문제 해결하기¶

원시 바이트¶

파일 이름 <커넥터_이름>/<테이블_이름>/<파티션>/offset_(<키>/<값>_)<타임스탬프>.gz 에는 원시 바이트에서 소스 파일 형식으로 변환되지 않은 레코드의 정확한 오프셋이 포함됩니다. 문제를 해결하려면 Kafka 커넥터에 레코드를 새 레코드로 다시 전송합니다.

원본 파일 형식(Avro, JSON 또는 프로토콜 버퍼)¶

Snowpipe가 Kafka 항목에 대해 생성된 내부 스테이지의 파일에서 데이터를 로드할 수 없는 경우 Kafka 커넥터는 소스 파일 형식의 대상 테이블에 대한 스테이지로 파일을 이동합니다.

파일 세트에서 문제가 여러 개 발생한 경우 COPY_HISTORY 출력의 FIRST_ERROR_MESSAGE 열은 발생한 첫 번째 오류만 나타냅니다. 파일의 모든 오류를 확인하려면 테이블 스테이지에서 파일을 검색하여 명명된 스테이지로 업로드한 후 VALIDATION_MODE 문을 실행하여 COPY INTO <테이블> 복사 옵션을 RETURN_ALL_ERRORS 로 설정합니다. VALIDATION_MODE 복사 옵션은 COPY 문에 로드할 데이터의 유효성을 검사하고 지정된 유효성 검사 옵션에 따라 결과를 반환하도록 지시합니다. 이 복사 옵션이 지정되면 데이터가 로드되지 않습니다. 이 문에서 Kafka 커넥터를 사용하여 로드하려고 시도한 파일 세트를 참조합니다.

데이터 파일의 문제가 해결되면 1개 이상의 COPY 문을 사용하여 데이터를 수동으로 로드할 수 있습니다.

다음 예에서는 mydb.public 데이터베이스 및 스키마의 mytable 테이블용 테이블 스테이지에 있는 데이터 파일을 참조합니다.

테이블 스테이지에서 데이터 파일의 유효성을 검사하고 오류를 해결하려면:

1. LIST 를 사용하여 테이블 스테이지에 있는 모든 파일을 나열합니다.

   예:

   LIST @mydb.public.%mytable;
   

   Copy

   이 섹션의 예시에서는 데이터 파일에 대한 소스 형식을 JSON으로 가정합니다.

2. GET 을 사용하여 Kafka 커넥터에서 생성한 파일을 로컬 컴퓨터에 다운로드합니다.

   예를 들어, 파일을 로컬 컴퓨터의 data 디렉터리에 다운로드합니다.

   Linux 또는 macOS:

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

   Copy

   Microsoft Windows:

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

   Copy

3. 소스 Kafka 파일과 동일한 형식으로 데이터 파일을 저장하는 CREATE STAGE 를 사용하여 명명된 내부 스테이지를 생성합니다.

   예를 들어, JSON 파일을 저장하는 이름이 kafka_json 인 내부 스테이지를 생성합니다.

   CREATE STAGE kafka_json FILE_FORMAT = (TYPE = JSON);
   

   Copy

4. PUT 을 사용하여 테이블 스테이지에서 다운로드한 파일을 업로드 합니다.

   예를 들어 다운로드한 파일을 로컬 컴퓨터의 data 디렉터리에 업로드합니다.

   Linux 또는 macOS:

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

   Copy

   Microsoft Windows:

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

   Copy

5. 테스트를 수행하기 위한 목적으로 베리언트 열 2개 포함된 임시 테이블을 생성합니다. 이 테이블은 스테이징된 데이터 파일의 유효성을 확인하기 위한 목적으로만 사용됩니다. 데이터가 테이블에 로드되지 않습니다. 현재 사용자 세션이 종료되면 테이블이 자동으로 삭제됩니다.

   CREATE TEMPORARY TABLE t1 (col1 variant);
   

   Copy

6. COPY INTO *테이블* … VALIDATION_MODE = ‘RETURN_ALL_ERRORS’ 문을 실행하여 데이터 파일에서 발생한 모든 오류를 검색합니다. 이 문은 지정된 스테이지에서 파일의 유효성을 검사합니다. 데이터가 테이블에 로드되지 않습니다.

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

   Copy

7. 로컬 컴퓨터의 데이터 파일에서 보고된 모든 오류를 수정합니다.

8. PUT 을 사용하여 수정된 파일을 테이블 스테이지 또는 명명된 내부 스테이지에 업로드합니다.

   다음 예시에서는 파일을 테이블 스테이지로 업로드하며, 다음과 같이 기존 파일을 덮어씁니다.

   Linux 또는 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. VALIDATION_MODE 옵션을 제외하고 COPY INTO 테이블 을 사용하여 대상 테이블에 데이터를 로드합니다.

   선택적으로 PURGE = TRUE 복사 옵션을 사용하여 데이터가 로드되면 스테이지에서 데이터 파일을 삭제하거나 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