Sink Connector Post Processors

사용자 지정 문서 ID 전략 만들기

기본 제공 문서 ID 추가 전략이 사용 사례에 맞지 않는 경우, 아래 단계에 따라 사용자 지정 문서 ID 전략을 정의할 수 있습니다.

1. 인터페이스 Id 전략을 구현하고 사용자 지정 구성 논리를 포함하는 Java 클래스를 만듭니다.

2. 클래스를 JAR 파일로 컴파일합니다.

3. 컴파일된 JAR을 모든 Kafka 작업자의 클래스 경로/플러그인 경로에 추가합니다. 플러그인 경로에 대한 자세한 내용은 Confluent 문서를 참조하세요.

4. document.id.strategy 설정을 모든 Kafka 워커에서 사용자 지정 클래스의 전체 클래스 이름으로 업데이트합니다.

IdStrategy 인터페이스의 구현 예를 보려면 커넥터와 함께 패키징된 id 전략 구현이 포함된 소스 코드 디렉터리를 참조하세요.

허용 목록 와일드카드 예시

* 와일드카드를 사용하여 여러 필드 이름과 일치시킬 수 있습니다. 다음 예시 구성은 다음 필드와 일치합니다.

• 이름이 "city"인 최상위 필드

• 이름이 "average"인 필드는 "wind_speed"라는 이름으로 시작하는 최상위 필드의 하위 문서입니다.

post.processor.chain=com.mongodb.kafka.connect.sink.processor.AllowListValueProjector
value.projection.type=AllowList
value.projection.list=city,wind_speed*.average

포스트 프로세서가 허용 목록 프로젝션을 적용한 후 다음 레코드를 출력합니다.

{
  "city": "Springfield",
  "wind_speed_10m": {
    "average": 3,
  },
  "wind_speed_80m": {
    "average": 8,
  }
}

와일드카드를 지정한 수준부터 시작하여 모든 수준의 객체와 일치하는 ** 와일드카드를 사용할 수 있습니다. 다음 와일드카드 매칭 예시에서는 이름이 "low"인 필드를 포함하는 모든 문서를 프로젝션합니다.

post.processor.chain=com.mongodb.kafka.connect.sink.processor.AllowListValueProjector
value.projection.type=AllowList
value.projection.list=**.low

프로젝션을 적용하는 포스트 프로세서는 다음 레코드를 출력합니다.

{
  "temperature": {
    "high": 28,
    "low": 24,
    "units": "C"
  },
  "soil_conditions": {
    "temperature": {
      "high": 22,
      "low": 17,
      "units": "C"
    }
  }
}

차단 목록 와일드카드 예시

와일드카드 패턴을 사용하여 다음 차단 목록 구성 예와 같이 특정 문서 수준의 필드를 일치시킬 수 있습니다.

post.processor.chain=com.mongodb.kafka.connect.sink.processor.BlockListValueProjector
value.projection.type=BlockList
value.projection.list=*.*.temperature

{
  "city": "Springfield",
  "temperature": {
    "high": 28,
    "low": 24,
    "units": "C"
  },
  "wind_speed_10m": {
    "average": 3,
    "units": "km/h"
  },
  "wind_speed_80m": {
    "average": 8,
    "units": "km/h"
  },
  "soil_conditions": {
    "moisture": {
      "average": 340,
      "units": "mm"
    }
  }
}

매핑으로 이름 변경 예시

RenameByMapping 포스트 프로세서 설정은 문자열과 일치하는 필드를 새 이름에 할당하는 하나 이상의 JSON 객체를 지정합니다. 각 객체에는 아래 표에 설명된 대로 oldName 요소에서 일치시킬 텍스트와 newName 요소의 대체 텍스트가 들어 있습니다.

다음 예시 속성은 키 문서의 "location" 필드와 일치하고 이름을 "country"로 바꿉니다.

field.renamer.mapping=[{"oldName":"key.location", "newName":"country"}]

이 설정은 RenameByMapping 포스트 프로세서가 원본 키 문서를 다음 문서로 변환하도록 지시합니다.

{
  "country": "Provence",
  "date_month": "October",
  "date_day": 17
}

다음과 같이 oldName 필드에 필드 이름이 추가된 값 문서를 지정하여 값 문서에 대해 유사한 필드 이름 할당을 수행할 수 있습니다.

field.renamer.mapping=[{"oldName":"value.flapjacks", "newName":"crepes"}]

이 설정은 RenameByMapping 포스트 프로세서에 원본 값 문서를 다음 문서로 변환하도록 지시합니다.

{
  "crepes": {
    "purchased": 598,
    "size": "large"
  }
}

다음 설정과 같이 문자열 형식의 JSON 배열을 사용하여 field.renamer.mapping 속성에 매핑을 하나 이상 지정할 수도 있습니다.

field.renamer.mapping=[{ "oldName":"key.location", "newName":"city" }, { "oldName":"value.crepes", "newName":"flapjacks" }]

정규 표현식으로 이름 바꾸기

RenameByRegex 포스트 프로세서 설정은 일치해야 하는 필드 이름과 텍스트 패턴, 일치하는 텍스트의 대체 값을 지정합니다. 다음 표에 설명된 필드를 포함하는 JSON 객체에서 이름 변경 표현식을 하나 이상 지정할 수 있습니다.

다음 예시 설정은 포스트 프로세서에 다음을 수행하도록 지시합니다.

• "date"로 시작하는 키 문서의 필드 이름을 일치시킵니다. 일치하는 필드 세트에서 _ 패턴과 일치하는 모든 텍스트를 - 문자로 바꿉니다.

• crepes의 하위 문서인 값 문서의 모든 필드 이름을 매칭합니다. 매칭 필드 세트에서 purchased 패턴과 일치하는 모든 텍스트를 quantity로 교체합니다.

field.renamer.regexp=[{"regexp":"^key\\.date.*$","pattern":"_","replace":"-"}, {"regexp":"^value\\.crepes\\..*","pattern":"purchased","replace":"quantity"}]

커넥터가 키 예시 문서와 값 예시 문서에 포스트 프로세서를 적용하면 다음과 같이 출력됩니다.

키 문서

{
  "location": "Provence",
  "date-month": "October",
  "date-day": 17
}

값 문서

{
  "crepes": {
    "quantity": 598,
    "size": "large"
  }
}