RHEL에 Debezium 설치

사전 요구 사항

• 소스 데이터베이스가 배포되고 Debezium 커넥터가 데이터베이스에 액세스할 수 있습니다.

• 커넥터가 소스 데이터베이스에 액세스하는 데 필요한 다음 정보를 알고 있습니다.

  • 데이터베이스 호스트의 이름 또는 IP 주소입니다.
  • 데이터베이스에 연결하기 위한 포트 번호입니다.
  • 커넥터가 데이터베이스에 로그인하는 데 사용할 수 있는 계정의 이름입니다.
  • 데이터베이스 사용자 계정의 암호입니다.
  • 데이터베이스 이름입니다.
  • 커넥터가 정보를 캡처할 테이블의 이름입니다.
  • 커넥터가 변경 이벤트를 발송할 Kafka 브로커의 이름입니다.
  • 커넥터가 데이터베이스 기록 정보를 보낼 Kafka 항목의 이름입니다.

절차

• JSON 형식의 Debezium 커넥터에 적용할 구성을 지정합니다.
  

  다음 예제에서는 Debezium MySQL 커넥터에 대한 간단한 구성을 보여줍니다.

  {
    "name": "inventory-connector",  

  1

  
    "config": {
      "connector.class": "io.debezium.connector.mysql.MySqlConnector", 

  2

  
      "tasks.max": "1",  

  3

  
      "database.hostname": "mysql",  

  4

  
      "database.port": "3306", 

  5

  
      "database.user": "debezium", 

  6

  
      "database.password": "dbz", 

  7

  
      "database.server.id": "184054",  

  8

  
      "topic.prefix": "dbserver1",   

  9

  
      "table.include.list": "public.inventory",  

  10

  
      "schema.history.internal.kafka.bootstrap.servers": "kafka:9092",  

  11

  
      "schema.history.internal.kafka.topic": "dbhistory.inventory"  

  12

  
    }
  }

  Copy to Clipboard Toggle word wrap

  1

  Kafka Connect 클러스터에 등록할 커넥터의 이름입니다.

  2

  커넥터 클래스의 이름입니다.

  3

  동시에 작동할 수 있는 작업 수입니다. 한 번에 하나의 작업만 작동해야 합니다.

  4

  호스트 데이터베이스 인스턴스의 호스트 이름 또는 IP 주소입니다.

  5

  데이터베이스 인스턴스의 포트 번호입니다.

  6

  Debezium이 데이터베이스에 연결하는 사용자 계정의 이름입니다.

  7

  데이터베이스 사용자 계정의 암호입니다.

  8

  커넥터의 고유한 숫자 ID입니다.
  이 속성은 MySQL 커넥터에만 사용됩니다.

  9

  커넥터가 변경 사항을 캡처하는 데이터베이스 서버 또는 서버 클러스터에 대한 논리 식별자 역할을 하는 문자열입니다. 지정된 문자열이 네임스페이스를 지정합니다. Debezium은 Avro 변환기를 사용할 때 커넥터가 쓰는 각 Kafka 항목에 이 이름을 접두사로 지정하고 Kafka Connect 스키마의 이름 및 해당 Avro 스키마의 네임스페이스를 추가합니다.

  10

  커넥터가 변경 이벤트를 캡처하는 테이블 목록입니다.

  11

  커넥터가 데이터베이스 스키마 기록을 전송하는 Kafka 브로커의 이름입니다. 지정된 브로커는 커넥터가 내보내는 변경 이벤트도 수신합니다.

  12

  스키마 기록을 저장하는 Kafka 항목의 이름입니다.
  커넥터가 다시 시작되면 커넥터가 중지된 시점에서 데이터베이스 로그 읽기를 다시 시작하여 오프라인 상태에서 발생한 모든 트랜잭션에 대한 이벤트를 출력합니다. 커넥터가 Kafka에 읽지 않은 트랜잭션의 변경 이벤트를 쓰기 전에 스키마 기록을 확인한 다음 원래 트랜잭션이 발생했을 때 적용되는 스키마를 적용합니다.

추가 정보

• 각 커넥터 유형에 대해 설정할 수 있는 구성 속성에 대한 자세한 내용은 Debezium 사용자 가이드 의 커넥터 배포 설명서를 참조하십시오.

사전 요구 사항

• Debezium을 배포할 호스트 환경은 지원되는 구성에서 Red Hat Enterprise Linux, AMQ Streams 및 Java를 실행합니다.
  

  • AMQ Streams 설치 방법에 대한 자세한 내용은 AMQ Streams 설치를 참조하십시오. https://access.redhat.com/documentation/en-us/red_hat_amq_streams/2.3/html-single/using_amq_streams_on_rhel/index#proc-installing-amq-streams-str
  • 단일 ZooKeeper 노드와 단일 Kafka 노드가 포함된 기본 비프로덕션 스트림 클러스터를 설치하는 방법에 대한 자세한 내용은 단일 노드 AMQ Streams 클러스터 실행을 참조하십시오.

• 호스트에 대한 관리 권한(sudo 액세스)이 있습니다.
• Apache ZooKeeper 및 Apache Kafka 브로커가 실행 중입니다.
• Kafka Connect는 독립 실행형 모드가 아닌 분산 모드로 실행됩니다.
• AMQ Streams를 설치할 때 생성된 kafka 사용자의 인증 정보를 알고 있습니다.
• 소스 데이터베이스가 배포되고 Debezium을 배포하는 호스트는 데이터베이스에 액세스할 수 있습니다.
• 커넥터를 구성하는 방법을 알고 있습니다.

추가 리소스

• 커넥터 플러그인 추가

사전 요구 사항

• 2.3절. “Red Hat Enterprise Linux에서 AMQ Streams를 사용하여 Debezium 배포” 의 지침에 따라 Red Hat Enterprise Linux에 커넥터를 배포했습니다. .Procedure

  1. 호스트의 터미널 창에서 다음 명령을 입력하여 Kafka Connect API에서 커넥터 목록을 요청합니다.

     $ curl -H "Accept:application/json" localhost:8083/connectors/

     Copy to Clipboard Toggle word wrap

     쿼리는 배포된 커넥터의 이름을 반환합니다. 예를 들면 다음과 같습니다.

     ["inventory-connector"]

     Copy to Clipboard Toggle word wrap

  2. 호스트의 터미널 창에서 다음 명령을 입력하여 커넥터가 실행 중인 작업을 확인합니다.

     $ curl -i -X GET -H "Accept:application/json" localhost:8083/connectors/inventory-connector

     Copy to Clipboard Toggle word wrap

     이 명령은 다음 예와 유사한 출력을 반환합니다.

     HTTP/1.1 200 OK
     Date: Thu, 06 Feb 2020 22:12:03 GMT
     Content-Type: application/json
     Content-Length: 531
     Server: Jetty(9.4.20.v20190813)
     
     {
       "name": "inventory-connector",
       ...
       "tasks": [
         {
           "connector": "inventory-connector",
           "task": 0
         }
       ]
     }

     Copy to Clipboard Toggle word wrap

  3. Kafka 클러스터의 주제 목록을 표시합니다.
     터미널 창에서 /opt/kafka/bin/ 로 이동하여 다음 쉘 스크립트를 실행합니다.

     ./kafka-topics.sh --bootstrap-server=localhost:9092 --list

     Copy to Clipboard Toggle word wrap

     Kafka 브로커는 커넥터가 생성하는 주제 목록을 반환합니다. 사용 가능한 주제는 커넥터의 snapshot.mode,snapshot.include.collection.list 설정 및 table.include.list 구성 속성에 따라 다릅니다. 기본적으로 커넥터는 데이터베이스의 시스템 이외의 각 테이블에 대한 항목을 만듭니다.

  4. 주제의 내용을 봅니다.
     터미널 창에서 /opt/kafka/bin/ 로 이동하여 kafka-console-consumer.sh 쉘 스크립트를 실행하여 이전 명령에서 반환한 주제 중 하나의 내용을 표시합니다.
     

     예를 들면 다음과 같습니다.

     ./kafka-console-consumer.sh \
     >     --bootstrap-server localhost:9092 \
     >     --from-beginning \
     >     --property print.key=true \
     >     --topic=dbserver1.inventory.products_on_hand

     Copy to Clipboard Toggle word wrap

     주제의 각 이벤트에 대해 명령에서 다음 출력과 유사한 정보를 반환합니다.

     예 2.1. Integration 변경 이벤트의 내용

     {"schema":{"type":"struct","fields":[{"type":"int32","optional":false,"field":"product_id"}],"optional":false,"name":"dbserver1.inventory.products_on_hand.Key"},"payload":{"product_id":101}}	{"schema":{"type":"struct","fields":[{"type":"struct","fields":[{"type":"int32","optional":false,"field":"product_id"},{"type":"int32","optional":false,"field":"quantity"}],"optional":true,"name":"dbserver1.inventory.products_on_hand.Value","field":"before"},{"type":"struct","fields":[{"type":"int32","optional":false,"field":"product_id"},{"type":"int32","optional":false,"field":"quantity"}],"optional":true,"name":"dbserver1.inventory.products_on_hand.Value","field":"after"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"version"},{"type":"string","optional":false,"field":"connector"},{"type":"string","optional":false,"field":"name"},{"type":"int64","optional":false,"field":"ts_ms"},{"type":"string","optional":true,"name":"io.debezium.data.Enum","version":1,"parameters":{"allowed":"true,last,false"},"default":"false","field":"snapshot"},{"type":"string","optional":false,"field":"db"},{"type":"string","optional":true,"field":"sequence"},{"type":"string","optional":true,"field":"table"},{"type":"int64","optional":false,"field":"server_id"},{"type":"string","optional":true,"field":"gtid"},{"type":"string","optional":false,"field":"file"},{"type":"int64","optional":false,"field":"pos"},{"type":"int32","optional":false,"field":"row"},{"type":"int64","optional":true,"field":"thread"},{"type":"string","optional":true,"field":"query"}],"optional":false,"name":"io.debezium.connector.mysql.Source","field":"source"},{"type":"string","optional":false,"field":"op"},{"type":"int64","optional":true,"field":"ts_ms"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"id"},{"type":"int64","optional":false,"field":"total_order"},{"type":"int64","optional":false,"field":"data_collection_order"}],"optional":true,"field":"transaction"}],"optional":false,"name":"dbserver1.inventory.products_on_hand.Envelope"},"payload":{"before":null,"after":{"product_id":101,"quantity":3},"source":{"version":"2.1.4.Final-redhat-00001","connector":"mysql","name":"inventory_connector_mysql","ts_ms":1638985247805,"snapshot":"true","db":"inventory","sequence":null,"table":"products_on_hand","server_id":0,"gtid":null,"file":"mysql-bin.000003","pos":156,"row":0,"thread":null,"query":null},"op":"r","ts_ms":1638985247805,"transaction":null}}

     Copy to Clipboard Toggle word wrap

     위 예제에서 페이로드 값은 커넥터 스냅샷이 테이블 inventory.ECDHE_on_hand 에서 읽기("op" ="r") 이벤트를 생성했음을 보여줍니다. product_id 레코드의 "before" 상태는 레코드에 이전 값이 없음을 나타내는 null 입니다. "after" 상태는 product_id 101 이 있는 항목에 대한 수량 을 표시합니다.