모델 전략 작성

비즈니스 키 지정하기

비즈니스 키는 하나 이상의 고유 식별 싱크 기록으로 구성된 값입니다. 싱크 커넥터는 싱크 기록의 _id 필드를 사용하기본적으로 여 비즈니스 키를 조회합니다. 다른 비즈니스 키를 지정하려면 사용자 정의 값을 사용하도록 Document Id Adder를 구성합니다.

다음 속성 예시에 표시된 것처럼 싱크 기록 키에서 _id 필드를 설정하도록 Document Id Adder를 구성할 수 있습니다.

document.id.strategy=com.mongodb.kafka.connect.sink.processor.id.strategy.PartialKeyStrategy
document.id.strategy.partial.key.projection.list=<comma-separated field names>
document.id.strategy.partial.key.projection.type=AllowList

또는 다음 속성 예시에 표시된 대로 싱크 기록 값에서 _id 필드를 설정하도록 구성할 수 있습니다.

document.id.strategy=com.mongodb.kafka.connect.sink.processor.id.strategy.PartialValueStrategy
document.id.strategy.partial.value.projection.list=<comma-separated field names>
document.id.strategy.partial.value.projection.type=AllowList

다음 쓰기 모델 전략에는 비즈니스 키가 필요합니다.

• ReplaceOneBusinessKeyStrategy

• DeleteOneBusinessKeyStrategy

• UpdateOneBusinessKeyTimestampStrategy

문서 ID 추가 포스트 프로세서 관련 세부 사항은 Document Id Adder Post Processor 구성하기를 참조하십시오.

하나의 타임스탬프 업데이트 전략

MongoDB에 문서를 작성할 때 타임스탬프를 추가하고 업데이트하도록 하나의 타임스탬프 업데이트 전략을 구성할 수 있습니다. 이 전략은 다음 조치를 수행합니다.

• 새 MongoDB 문서를 삽입할 때는 커넥터가 _insertedTS, _modifiedTS 필드를 커넥터 서버의 현재 시간으로 설정합니다.

• 커넥터가 기존 MongoDB 문서를 업데이트하면 _modifiedTS 필드는 커넥터 서버의 현재 시간으로 업데이트됩니다.

경로를 따라 열차의 위치를 추적하려고 하는 상황에서 싱크 커넥터가 다음과 같은 구조의 메시지를 수신한다고 가정해 보겠습니다.

{
  "_id": "MN-1234",
  "start": "Beacon",
  "destination": "Grand Central"
  "position": [ 40, -73 ]
}

커넥터가 메시지의 _id 값을 사용하여 MongoDB 문서의 _id 필드를 할당하도록 지정하려면 ProvidedInValueStrategy를 사용하십시오. 아이디를 지정하고 다음과 같이 모델 전략 속성을 작성합니다.

document.id.strategy=com.mongodb.kafka.connect.sink.processor.id.strategy.ProvidedInValueStrategy
writemodel.strategy=com.mongodb.kafka.connect.sink.writemodel.strategy.UpdateOneTimestampsStrategy

싱크 커넥터는 위의 기록 예시를 처리한 후 다음 문서와 같이 _insertedTS 및 _modifiedTS 필드가 포함된 문서를 삽입합니다.

{
  "_id": "MN-1234",
  "_insertedTS": ISODate("2021-09-20T15:08:000Z"),
  "_modifiedTS": ISODate("2021-09-20T15:08:000Z"),
  "start": "Beacon",
  "destination": "Grand Central"
  "position": [ 40, -73 ]
}

한 시간 후에 열차는 다음 기록에 표시된 대로 경로를 따라 새 위치를 새로운 위치로 보고합니다.

{
  "_id": "MN-1234",
  "start": "Beacon",
  "destination": "Grand Central"
  "position": [ 42, -75 ]
}

Sink Connector가 이전 기록을 처리하고 나면 다음 데이터가 포함된 문서를 삽입합니다.

{
  "_id": "MN-1234",
  "_insertedTS": ISODate("2021-09-20T15:08:000Z"),
  "_modifiedTS": ISODate("2021-09-20T16:08:000Z"),
  "start": "Beacon",
  "destination": "Grand Central"
  "position": [ 42, -75 ]
}

ProvidedInValueStrategy 관련 정보는 Document Id Adder Post Processor 구성하기를 참조하십시오.

하나의 비즈니스 키 교체 전략

비즈니스 키 값과 일치하는 문서를 교체하도록 하나의 비즈니스 키 교체 전략을 구성할 수 있습니다. 기록의 여러 필드에 비즈니스 키를 정의하고 매칭 비즈니스 키가 포함된 문서를 교체하도록 커넥터를 구성하기 위해 다음 작업을 수행합니다.

1. 컬렉션에서 비즈니스 키 필드에 해당하는 고유 인덱스를 생성합니다.

2. 커넥터 구성에서 비즈니스 키에 속하는 필드를 식별하기 위해 PartialValueStrategy ID 전략을 지정합니다.

3. 커넥터 구성에서 ReplaceOneBusinessKeyStrategy 쓰기 모델 전략을 지정합니다.

flight_no와 airport_code로 각각 표시된 항공편 번호와 공항 위치를 기준으로 항공기 용량을 추적한다고 가정해 봅시다. 메시지 예시는 다음 정보를 포함합니다.

{
  "flight_no": "Z342",
  "airport_code": "LAX",
  "seats": {
    "capacity": 180,
    "occupied": 152
  }
}

flight_no와 airport_code를 비즈니스 키로 사용하여 전략을 구현하기 위해서는 우선 MongoDB Shell에서 해당 필드에 고유 인덱스를 생성합니다.

db.collection.createIndex({ "flight_no": 1, "airport_code": 1}, { unique: true })

그 다음 프로젝션 목록에서 PartialValueStrategy 전략 및 비즈니스 키 필드를 지정합니다. 다음과 같이 ID를 지정하고 모델 전략 구성을 작성합니다.

document.id.strategy=com.mongodb.kafka.connect.sink.processor.id.strategy.PartialValueStrategy
document.id.strategy.partial.value.projection.list=flight_no,airport_code
document.id.strategy.partial.value.projection.type=AllowList
writemodel.strategy=com.mongodb.kafka.connect.sink.writemodel.strategy.ReplaceOneBusinessKeyStrategy

컬렉션에 삽입된 샘플 데이터 포함 사항:

{
  "flight_no": "Z342"
  "airport_code": "LAX",
  "seats": {
    "capacity": 180,
    "occupied": 152
  }
}

커넥터가 기존 문서의 비즈니스 키와 일치하는 싱크 데이터를 처리할 때는 비즈니스 키 필드를 변경하지 않고 문서를 새 값으로 교체합니다.

{
  "flight_no": "Z342"
  "airport_code": "LAX",
  "status": "canceled"
}

싱크 데이터를 처리한 후에는 커넥터가 MongoDB의 원본 샘플 문서를 이전 샘플 문서로 교체합니다.

하나의 비즈니스 키 삭제 전략

하나의 비즈니스 키 삭제 전략을 사용하여 비즈니스 키와 일치하는 메시지를 수신할 때 문서를 제거하도록 커넥터를 구성할 수 있습니다. 기록의 여러 필드에서 비즈니스 키를 설정하고 일치하는 비즈니스 키가 포함된 문서를 삭제하도록 커넥터를 구성하려면 다음 작업을 수행합니다.

1. MongoDB 컬렉션에 대하여 비즈니스 키 필드에 해당하는 고유 인덱스를 생성합니다.

2. 커넥터 구성에서 비즈니스 키에 속하는 필드를 식별하기 위한 식별 전략으로 PartialValueStrategy를 지정합니다.

3. 커넥터 구성에서 DeleteOneBusinessKeyStrategy 쓰기 모델 전략을 지정합니다.

다음과 유사한 문서가 포함된 컬렉션에서 특정 연도의 캘린더 이벤트를 삭제한다고 가정해 보겠습니다.

{
  "year": 2005,
  "month": 3,
  "day": 15,
  "event": "Dentist Appointment"
}

year를 비즈니스 키로 사용하여 전략을 구현하기 위해서는 먼저 MongoDB Shell에서 다음 필드에 고유 인덱스를 생성합니다.

db.collection.createIndex({ "year": 1 }, { unique: true })

다음으로 비즈니스 키를 지정하고 다음과 같이 구성에 모델 전략을 작성합니다.

document.id.strategy=com.mongodb.kafka.connect.sink.processor.id.strategy.PartialValueStrategy
document.id.strategy.partial.value.projection.list=year
document.id.strategy.partial.value.projection.type=AllowList
writemodel.strategy=com.mongodb.kafka.connect.sink.writemodel.strategy.DeleteOneBusinessKeyStrategy

커넥터가 비즈니스 키 year가 포함된 싱크 기록을 처리하는 경우, MongoDB가 반환한 매칭 필드 값을 가진 첫 번째 문서를 삭제합니다. 커넥터가 다음 값 데이터가 포함된 싱크 기록을 처리한다고 가정해 봅시다.

{
  "year": 2005,
  ...
}

커넥터가 이전 기록을 처리할 때 원본 '치과 예약' 샘플 문서처럼 '2005'라는 값을 가진 year 필드를 포함하는 첫 번째 문서를 컬렉션에서 삭제합니다.

Sample Write Model 전략

다음 사용자 지정 쓰기 모델 전략은 싱크 기록의 _id 필드와 일치하는 MongoDB 문서를 싱크 기록의 fullDocument 필드 값으로 교체하는 쓰기 작업을 반환합니다.

/**
 * Custom write model strategy
 *
 * This class reads the 'fullDocument' field from a change stream and
 * returns a ReplaceOne operation.
 */
public class CustomWriteModelStrategy implements WriteModelStrategy {
  private static String ID = "_id";
  @Override
  public WriteModel<BsonDocument> createWriteModel(final SinkDocument document) {
    BsonDocument changeStreamDocument = document.getValueDoc()
        .orElseThrow(() -> new DataException("Missing value document"));
    BsonDocument fullDocument = changeStreamDocument.getDocument("fullDocument", new BsonDocument());
    if (fullDocument.isEmpty()) {
      return null; // Return null to indicate no op.
    }
    return new ReplaceOneModel<>(Filters.eq(ID, fullDocument.get(ID)), fullDocument);
  }
}

사용자 지정 쓰기 모델 전략의 또 다른 예는 GitHub의 UpsertAsPartOfDocumentStrategy 예시 전략을 참조하십시오.

전략 설치 방법

사용자 지정 쓰기 전략을 사용하도록 싱크 커넥터를 구성하려면 다음 조치를 완료해야 합니다.

1. 사용자 지정 쓰기 전략 클래스를 JAR 파일로 컴파일합니다.

2. 컴파일된 JAR을 Kafka 작업자의 클래스패스/플러그인 패스에 추가하세요. 플러그인 패스에 대한 자세한 내용은 Confluent 설명서를 참조하세요.

   참고

   Kafka Connect는 플러그인을 개별적으로 로드합니다. 사용자 지정 쓰기 전략을 배포하는 경우 커넥터 JAR과 쓰기 모델 전략 JAR이 모두 동일한 경로에 있어야 합니다. 다음 경로와 유사한 경로를 사용하십시오.

   <plugin.path>/mongo-kafka-connect/mongo-kafka-connect-all.jar

   <plugin.path>/mongo-kafka-connect/custom-write-model-strategy.jar

   Kafka Connect 플러그인에 대해 자세히 알아보려면 Confluent에서 제공하는 이 가이드를 참조하세요.

3. writemodel.strategy 구성 설정에서 사용자 지정 클래스를 지정합니다.

클래스를 JAR 파일로 컴파일하는 방법은 Java SE 문서의 JAR 배포 가이드를 참조하십시오.