Opensearch sink connector(feat. MSK)

Streamlining MSK Connect to OpenSearch for Seamless Loading — Part A

Part A — Simplifying Data Loading with MSK Connector for OpenSearch

https://medium.com/ankercloud-engineering/streamlining-msk-connect-to-opensearch-for-seamless-loading-1c563b94fa2b

releases

v1.5.0

위 두 개의 파일을 모두 다운 받아야 합니다

사용자 정의 플러그인 생성

두 커넥터를 다운로드하고 압축을 풀면 “transforms-for-apache-kafka-connect-1.5.0/transforms-for-apache-kafka-connect-1.5.0” 폴더로 이동하십시오. 여기에는 두 개의 .jar 파일이 있습니다. “transforms-for-apache-kafka-connect-1.5.0.jar”을 “opensearch-connector-for-apache-kafka-3.1.1/opensearch-connector-for-apache-kafka-3.1.1” 폴더로 복사하십시오.

“slf4j-api-1.7.36.jar”은 이미 “opensearch-connector-for-apache-kafka-3.1.1” 폴더에 있으므로 복사할 필요가 없습니다.

이 작업이 완료되면 “opensearch-connector-for-apache-kafka-3.1.1” 폴더를 압축하고, 우리의 사용자 정의 플러그인이 생성됩니다.

GitHub - Aiven-Open/opensearch-connector-for-apache-kafka: Aiven's OpenSearch® Connector for Apache Kafka®

Aiven's OpenSearch® Connector for Apache Kafka®. Contribute to Aiven-Open/opensearch-connector-for-apache-kafka development by creating an account on GitHub.

https://github.com/Aiven-Open/opensearch-connector-for-apache-kafka?tab=readme-ov-file

zip -r opensearch-connector-for-apache-kafka-3.1.1.zip opensearch-connector-for-apache-kafka-3.1.1
Shell
복사

위의 깃허브 내용대로 압축을 풀어서, plugin 폴더 안에 배치합니다.

압축 해제된 바이너리를 각 Connect 작업 노드의 디렉토리에 배치합니다.

예를 들어, /connectors

이 경우, opensearch-connector-for-kafka를 connectors 폴더에 배치합니다

connectors
└── opensearch-connector-for-apache-kafka
Plain Text
복사

그런 다음, 각 연결 작업자 구성에서 /connectors을 plugin.path 구성에 추가하십시오:

plugin.path=/connectors

 $ tree connectors/
 
 connectors/
 ├── msk-datagen
 │   └── msk-data-generator.jar
 ├── opensearch-connector
 
 ...
│   ├── transforms-for-apache-kafka-connect-1.5.0.jar // 추가한 파일
│   ├── opensearch-2.6.0.jar
│   ├── opensearch-cli-2.6.0.jar
│   ├── opensearch-common-2.6.0.jar
│   ├── opensearch-connector-for-apache-kafka-3.1.0.jar
│   ├── opensearch-core-2.6.0.jar

...
└── opensearch-connector.zip
Shell
복사

opensearch-connector 폴더를 압축해서 opensearch-connector.zip으로 만든 후,

MSK용 커스텀 커넥터 플러그인을 만들 예정이라면, "msk-custom-plugins"라는 이름의 S3 버킷을 만든 후, 우리의 zip 파일 "opensearch-connector-for-apache-kafka-3.1.1.zip"을 그 안에 업로드하면 됩니다.

MSK 커넥터 플러그인 생성 프로세스는 아래 글을 참고하시면 더 도움이 될 것입니다.

Streamlining MSK Connect to OpenSearch for Seamless Loading — Part A

Part A — Simplifying Data Loading with MSK Connector for OpenSearch

https://medium.com/ankercloud-engineering/streamlining-msk-connect-to-opensearch-for-seamless-loading-1c563b94fa2b

참고로 connectors 를 플러그인 path로 잡은 이후, MSK가 아닌 기존 카프카 커넥터를 재실행하면,

아래 명령어를 통해 플러그인 생성을 확인할 수 있습니다.

curl http://localhost:8083/connector-plugins | jq .

[
  ...
  {
    "class": "io.aiven.kafka.connect.opensearch.OpensearchSinkConnector",
    "type": "sink",
    "version": "3.0.0"
  },
  ...
]
Shell
복사

Kafka Connect for AWS Services Integration - Part 4 Develop Aiven OpenSearch Sink Connector

Kafka Connect can be an effective tool to ingest data from Apache Kafka into OpenSearch. In this post, we will discuss how to develop a data pipeline from Apache Kafka into OpenSearch locally using Docker while the pipeline will be deployed on AWS in the next post. Fake impressions and clicks data will be pushed into Kafka topics using a Kafka source connector and those records will be ingested into OpenSearch indexes using a sink connector for near-real time analytics.

https://jaehyeon.me/blog/2023-10-23-kafka-connect-for-aws-part-4/

Sink connector 생성

이제 플러그인을 만들었으니, sink connector를 생성할 차례입니다.

먼저 sink connector가 사용할 토픽을 만들어야 합니다.

위의 내용을 참고하여 sink connector json config(configs/source.json)를 작성해봅시다

{
  "name": "opensearch-sink",
  "config": {
    "connector.class": "io.aiven.kafka.connect.opensearch.OpensearchSinkConnector",
    "tasks.max": "2",
    "topics": "impressions",
    "key.converter": "org.apache.kafka.connect.storage.StringConverter",
    "key.converter.schemas.enable": false,
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": false,
    "connection.url": "https://opensearch-node1:9200",
    "connection.username": "admin",
    "connection.password": "admin",
    "schema.ignore": true,
    "key.ignore": true,
    "type.name": "_doc",
    "behavior.on.malformed.documents": "fail",
    "behavior.on.null.values": "ignore",
    "behavior.on.version.conflict": "ignore",
    "errors.deadletterqueue.topic.name": "ad-tech-dl",
    "errors.tolerance": "all",
    "errors.deadletterqueue.context.headers.enable": true,
    "errors.deadletterqueue.topic.replication.factor": 1,
    "transforms": "insertTS,formatTS",
    "transforms.insertTS.type": "org.apache.kafka.connect.transforms.InsertField$Value",
    "transforms.insertTS.timestamp.field": "created_at",
    "transforms.formatTS.type": "org.apache.kafka.connect.transforms.TimestampConverter$Value",
    "transforms.formatTS.format": "yyyy-MM-dd HH:mm:ss",
    "transforms.formatTS.field": "created_at",
    "transforms.formatTS.target.type": "string"
  }
}
Shell
복사

🎄 Twelve Days of SMT 🎄 - Day 8: TimestampConverter

The TimestampConverter Single Message Transform lets you work with timestamp fields in Kafka messages. You can convert a string into a native Timestamp type (or Date or Time), as well as Unix epoch - and the same in reverse too.

https://rmoff.net/2020/12/17/twelve-days-of-smt-day-8-timestampconverter/

설정 설명

위와 같이 싱크 커넥터 설정을 마쳤다면, 이제 아래 rest api를 통해 생성을 해봅시다.

$ curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" http://localhost:8083/connectors/ -d @configs/source.json
Shell
복사

아래와 같이 뜨면 성공입니다.

$ curl http://localhost:8083/connectors/ad-tech-sink/status | json_pp
{
   "connector" : {
      "state" : "RUNNING",
      "worker_id" : "172.19.0.8:8083"
   },
   "name" : "ad-tech-sink",
   "tasks" : [
      {
         "id" : 0,
         "state" : "RUNNING",
         "worker_id" : "172.19.0.8:8083"
      },
      {
         "id" : 1,
         "state" : "RUNNING",
         "worker_id" : "172.19.0.8:8083"
      }
   ],
   "type" : "source"
}
Shell
복사

참고로 data ingestion을 할 때, created dt와 같은 타임 스탬프를 sink시에 남기고 싶으면, SMT를 사용하면 됩니다. (이벤트 타임스탬프 속성은 시간적 분석을 수행하는 데 유용할 수 있습니다.)

원본 커넥터에서 이를 설정하는 종합적인 방법을 찾지 못하므로, 새로운 필드인 created_at이 단일 메시지 변환 (SMTs)을 사용하여 추가되었습니다. 구체적으로는 두 가지 변환, insertTS와 formatTS를 추가했습니다. 이름에서 알 수 있듯이 전자는 시스템 타임스탬프 값을 삽입하고 후자는 이를 yyyy-MM-dd HH:mm:ss 형식으로 포맷팅합니다.

"transforms": "insertTS,formatTS",
"transforms.insertTS.type": "org.apache.kafka.connect.transforms.InsertField$Value",
"transforms.insertTS.timestamp.field": "created_at",
"transforms.formatTS.type": "org.apache.kafka.connect.transforms.TimestampConverter$Value",
"transforms.formatTS.format": "yyyy-MM-dd HH:mm:ss",
"transforms.formatTS.field": "created_at",
"transforms.formatTS.target.type": "string"
Shell
복사

위와 같은 설정을 json config에 넣어서 sink connector를 업데이트하거나 재생성하면, 됩니다.