Avro 데이터 타입 수집 구성

Avro 데이터 타입 수집 구성

이 페이지에서는 스트리밍 커넥터(Kafka 고성능, Kinesis 고성능)가 기본 JsonTreeReader 대신 AvroReader를 사용해 Avro 인코딩 메시지를 수집하도록 전환하는 방법을 설명해요. 지원되는 스키마 접근 전략 4가지를 설정 단계별로 다룹니다.

출처: Snowflake 문서

본문

Note

이 커넥터는 Snowflake Connector Terms에 의해 규율됩니다.

스트리밍 커넥터(Kafka 고성능, Kinesis 고성능)는 Consume* 프로세서(ConsumeKafka / ConsumeKinesis)를 JsonTreeReader 컨트롤러 서비스와 함께 사용해 들어오는 메시지를 파싱합니다. 이 토픽은 JsonTreeReader를 AvroReader로 교체해 Avro 인코딩 메시지로 전환하는 방법을 설명합니다. 단계는 두 커넥터에 동일합니다 — Kinesis 커넥터에서는 ConsumeKafka가 언급된 곳에 ConsumeKinesis 프로세서를 사용하세요.

Tip

이 사용자 정의를 손으로 적용할 필요 없습니다. Snowflake CoCo의 Openflow 스킬이 대신 수행할 수 있습니다 — 원하는 변경을 설명하면 이 페이지의 단계에 따라 흐름을 편집합니다. 구성 요소를 수동으로 구성하는 대신 이 스킬을 사용하는 것을 권장합니다.

다음 스키마 접근 전략이 지원됩니다:

  • 인라인 스키마 텍스트 (Inline schema text) — 서비스 구성에 Avro 스키마를 직접 제공.

  • 임베디드 Avro 스키마 (Embedded Avro schema) — Avro 컨테이너 파일 헤더에서 스키마를 읽음. 외부 구성 불필요.

  • Confluent Schema Registry — 메시지에서 ConfluentEncodedSchemaReferenceReader로 스키마 ID를 읽고 ConfluentSchemaRegistry에 대해 해석.

  • AWS Glue Schema Registry — AWS Glue Schema Registry(AmazonGlueSchemaRegistry)에서 이름으로 스키마를 조회. 주로 Amazon MSK 고객이 사용.

전제 조건

  • Openflow에 배포된 기존 Kafka 고성능 또는 Kinesis 고성능 커넥터가 있어야 합니다.

  • Kafka 토픽 / Kinesis 스트림이 Avro 인코딩 메시지를 생성해야 합니다.

  • Confluent Schema Registry를 사용한다면: 레지스트리 URL과 Openflow 런타임에서 그 레지스트리로의 네트워크 접근이 있어야 합니다. Snowflake 관리 배포(SPCS)에서는 레지스트리로의 아웃바운드 연결을 허용하도록 적절한 External Access Integration이 구성되고 런타임에 할당되어 있어야 합니다.

Avro 데이터 타입 수집 구성

아래 단계는 Kafka 고성능과 Kinesis 고성능 커넥터에 동일하게 적용됩니다(Kinesis 커넥터에서는 ConsumeKafka가 언급된 곳에 ConsumeKinesis 프로세서를 사용).

1단계: AvroReader 컨트롤러 서비스 생성

  1. Openflow UI에서 커넥터의 프로세스 그룹을 엽니다.

  2. Configure > Controller Services(톱니바퀴 아이콘)로 이동합니다.

  3. **+**를 선택해 새 컨트롤러 서비스를 추가합니다.

  4. AvroReader(org.apache.nifi.avro.AvroReader)를 검색해 선택합니다.

  5. Add를 선택합니다.

2단계에서 선택한 스키마 접근 전략과 무관하게 AvroReader에는 다음 속성이 있습니다:

Property Description
Fast Reader Enabled 활성화하면 성능을 높이지만 메모리 사용량을 늘리는 최적화된 리더를 사용합니다. Avro 처리 중 OutOfMemoryError가 발생하면 비활성화하세요.
Cache Size 메모리에 캐시할 스키마 수. 스키마가 몇 개뿐이면 기본값으로 충분합니다. 단일 토픽/스트림이 높은 처리량으로 매우 많은 서로 다른 스키마를 나른다면, 스키마가 반복적으로 축출·재인출되지 않도록 값을 올리세요 — 더 높은 메모리 소비의 대가로.

2단계: AvroReader에서 스키마 접근 전략 구성

1단계에서 만든 AvroReader 컨트롤러 서비스에서 스키마 접근 전략을 구성합니다. Avro 스키마가 분산되는 방식과 일치하는 옵션을 선택하세요.

옵션 A — 인라인 스키마 텍스트 (Use 'Schema Text' Property)

Avro 스키마를 텍스트 문자열로 가지고 있고 런타임에 변하지 않을 때 사용하세요.

  1. AvroReader 서비스의 Edit(톱니바퀴) 아이콘을 선택합니다.

  2. 다음 속성을 설정합니다:

Property Value
Schema Access Strategy Use 'Schema Text' Property
Schema Text The full Avro-formatted schema text. Supports Expression Language.
  1. Apply를 선택합니다.

옵션 B — 임베디드 Avro 스키마 (Use Embedded Avro Schema)

메시지가 파일 헤더에 라이터 스키마를 포함하는 Avro Object Container Files(OCF)로 직렬화될 때 사용하세요. 추가 서비스가 필요 없습니다.

  1. AvroReader 서비스의 Edit(톱니바퀴) 아이콘을 선택합니다.

  2. 다음 속성을 설정합니다:

Property Value
Schema Access Strategy Use Embedded Avro Schema
  1. Apply를 선택합니다.

Note

이 전략은 자체 포함된 Avro 컨테이너 파일에서만 동작합니다. Kafka 메시지가 파일 헤더 없는 원시 Avro 인코딩 레코드라면(Kafka 프로듀서의 일반적인 경우) 옵션 A 또는 옵션 C를 대신 사용하세요.

옵션 C — Confluent Schema Registry (Schema Reference Reader)

메시지가 Confluent 와이어 형식(매직 바이트 0x00 뒤에 4바이트 스키마 ID)으로 인코딩될 때 사용하세요. 추가 컨트롤러 서비스 두 개가 필요합니다: ConfluentSchemaRegistry는 스키마 ID를 실제 스키마로 해석하고, ConfluentEncodedSchemaReferenceReader는 각 메시지에서 스키마 ID를 읽습니다.

ConfluentSchemaRegistry 컨트롤러 서비스 생성:

  1. Configure > Controller Services로 이동합니다.

  2. **+**를 선택하고 ConfluentSchemaRegistry(org.apache.nifi.confluent.schemaregistry.ConfluentSchemaRegistry)를 검색합니다.

  3. Add를 선택합니다.

  4. Edit(톱니바퀴) 아이콘을 선택하고 다음 속성을 설정합니다:

Property Value
Schema Registry URLs (required) Comma-separated URL(s) of your Confluent Schema Registry, for example, https://schema-registry.example.com:8081.
SSL Context Service (optional) An SSLContextService if the registry requires TLS. Implementations: StandardSSLContextService, StandardRestrictedSSLContextService, PEMEncodedSSLContextProvider.
Communications Timeout (required) How long to wait for a response from the registry before failing.
Cache Size (required) Number of schemas to cache locally. Raise it when a single topic/stream carries many distinct schemas at high throughput (more memory); the default is fine for a few schemas.
Cache Expiration (required) How long cached schemas are valid before being re-fetched.
Authentication Type NONE or BASIC if the registry requires HTTP Basic authentication.
Username Username for Basic authentication. Only used when Authentication Type is BASIC.
Password Password for Basic authentication. Sensitive property. Only used when Authentication Type is BASIC.

  1. Apply를 선택합니다.

  2. Enable(번개 아이콘)을 선택하고 상태가 Enabled로 표시될 때까지 기다립니다.

ConfluentEncodedSchemaReferenceReader 컨트롤러 서비스 생성:

  1. Configure > Controller Services로 이동합니다.

  2. **+**를 선택하고 ConfluentEncodedSchemaReferenceReader(org.apache.nifi.confluent.schemaregistry.ConfluentEncodedSchemaReferenceReader)를 검색합니다.

  3. Add를 선택합니다.

  4. Enable 아이콘을 선택하고 상태가 Enabled로 표시될 때까지 기다립니다.

Note

ConfluentEncodedSchemaReferenceReader에는 구성 가능한 속성이 없습니다. 각 메시지의 시작 부분에서 Confluent 인코딩 스키마 ID(매직 바이트 0x00 + 4바이트 정수)를 읽을 뿐입니다.

AvroReader가 스키마 참조 리더를 사용하도록 구성:

  1. 1단계에서 만든 AvroReader 서비스의 Edit(톱니바퀴) 아이콘을 선택합니다.

  2. 다음 속성을 설정합니다:

Property Value
Schema Access Strategy Schema Reference Reader
Schema Reference Reader Select the ConfluentEncodedSchemaReferenceReader created above.
Schema Registry Select the ConfluentSchemaRegistry created above.
  1. Apply를 선택합니다.

옵션 D — AWS Glue Schema Registry (Schema Name)

Amazon MSK 프로듀서가 AWS Glue Schema Registry에 Avro 스키마를 등록할 때 사용하세요. 추가 컨트롤러 서비스 하나가 필요합니다: AmazonGlueSchemaRegistry가 스키마를 이름으로 해석합니다.

AmazonGlueSchemaRegistry 컨트롤러 서비스 생성:

  1. Configure > Controller Services로 이동합니다.

  2. **+**를 선택하고 AmazonGlueSchemaRegistry를 검색합니다.

  3. Add를 선택합니다.

  4. Edit(톱니바퀴) 아이콘을 선택하고 다음 속성을 설정합니다:

Property Value
Schema Registry Name (required) The name of your Glue Schema Registry.
Region (required) The AWS region of the registry, for example, us-west-2.
AWS Credentials Provider Service Reuse the connector's AWSCredentialsProviderControllerService (Kinesis), or create one for MSK.
Cache Size (required) Number of schemas to cache locally. Raise for many distinct schemas at high throughput (more memory); the default is fine otherwise.
Cache Expiration (required) How long cached schemas are valid before being re-fetched.
  1. Apply를 선택한 뒤 서비스를 Enable합니다.

AmazonGlueEncodedSchemaReferenceReader 컨트롤러 서비스 생성:

  1. Configure > Controller Services로 이동합니다.

  2. **+**를 선택하고 AmazonGlueEncodedSchemaReferenceReader를 검색합니다.

  3. Add를 선택합니다.

  4. Enable 아이콘을 선택하고 상태가 Enabled로 표시될 때까지 기다립니다.

Note

AmazonGlueEncodedSchemaReferenceReader는 Glue Schema Registry 직렬화기가 각 메시지에 내장한 AWS Glue 스키마 참조(UUID와 버전 번호)를 읽습니다. 구성 가능한 속성이 없습니다.

AvroReader가 스키마 참조 리더를 사용하도록 구성:

  1. 1단계에서 만든 AvroReader 서비스의 Edit(톱니바퀴) 아이콘을 선택합니다.

  2. 다음 속성을 설정합니다:

Property Value
Schema Access Strategy Schema Reference Reader
Schema Reference Reader Select the AmazonGlueEncodedSchemaReferenceReader created above.
Schema Registry Select the AmazonGlueSchemaRegistry created above.
  1. Apply를 선택합니다.

3단계: AvroReader 컨트롤러 서비스 활성화

AvroReader의 Enable(번개 아이콘)을 선택하고 상태가 Enabled로 표시될 때까지 기다립니다.

4단계: 소스 프로세서 업데이트

  1. ConsumeKafka 프로세서(Kinesis 커넥터에서는 ConsumeKinesis)를 더블클릭해 속성을 엽니다.

  2. 다음 속성을 업데이트합니다:

Property Value
Record Reader Select the AvroReader created above.
  1. Apply를 선택합니다.

5단계: JsonTreeReader 컨트롤러 서비스 비활성화

원래 JsonTreeReader는 더 이상 필요하지 않습니다.

  1. 프로세스 그룹이 실행 중이면 중지합니다.

  2. Configure > Controller Services로 이동합니다.

  3. JsonTreeReader의 Disable 아이콘을 선택합니다.

  4. 서비스가 다른 프로세서에서 참조하지 않는다면 Delete(휴지통) 아이콘을 선택해 삭제할 수도 있습니다.

  5. 프로세스 그룹을 시작합니다.

문제 해결

Symptom Likely cause
SchemaNotFoundException at runtime The schema ID in the message is not present in the registry, or the registry URL is misconfigured. Verify Schema Registry URLs in ConfluentSchemaRegistry.
InvalidAvroSchemaException or parse failures The Schema Text does not match the actual message schema. Compare the schema in the service with the one used by the producer. Failed messages are routed to the parse-failure relationship (parse failure on Kafka, parse.failure on Kinesis).
ConfluentSchemaRegistry fails to enable Network connectivity issue between the Openflow runtime and the registry. Check the External Access Integration and that the registry URL is reachable from the cluster.
Authentication failure against registry Registry requires Basic auth — set Authentication Type to BASIC and provide Username and Password in ConfluentSchemaRegistry.

더 알아보기 (Learn more)