Chuyển tới nội dung chính

2.9 PostgreSQL Sink Connector

*Tạo connector, Type là sink, Database là PostgreSQL

Pre-condition: Status CDC service healthy

Cấu hình PostgreSQL - Cấp quyền trên schema​

GRANT USAGE ON SCHEMA public TO username;
 GRANT CREATE ON SCHEMA public TO username;

Các bước tạo connector:​

Bước 1: Tại thanh menu chọn Data Platform > chọn Workspace Management > chọn Workspace name

Bước 2: Tại phần My services chọn CDC service

Bước 3: Tại màn detail CDC service > Chọn tab Connectors > nhấn Create a connector

image-1

Bước 4: Nhập các thông tin màn Connector Information:

  • Name (required): tên connector

Chú ý: Tên connector có thể chứa các kí tự chữ cái thường a-z hoặc các kí tự số 0-9. Đặc biệt không dùng dấu cách có thể thay dấu cách bằng dấu “-”.

  • Type (required): chọn sink
  • Database (required): chọn PostgreSQL

image-2

Bước 5. Nhấn Next để chuyển qua màn Properties

Nhập thông tin màn Properties

  • Trường hợp chọn Manual configuration - Điền các thông tin:
    • Database name (required): chọn database
    • Host Name (required): Hostname hoặc IP của PostgreSQL
    • Port ( required): PostgreSQL server port, mặc định là: 5432.
    • Database name (required): Database đích mà Connector sẽ sink dữ liệu vào
    • Username (required): Username sử dụng bởi Connector
    • Password (required): Password sử dụng bởi Connector

image-3

  • Trường hợp chọn From Database Engine - Điền các thông tin:
    • Host Name (required): Hostname hoặc IP của PostgreSQL
    • Port (required): PostgreSQL server port, mặc định là: 5432.
    • Database name (required): Database đích mà Connector sẽ sink dữ liệu vào
    • Username (required): Username sử dụng bởi Connector
    • Password (required): Password sử dụng bởi Connector

image-4

Nhấn Test connection để kiểm tra kết nối từ Workspace tới Database đã nhập

  • Converter
    • Converter key: chọn giá trị key cho converter
    • Converter key schema enable: chọn giá trị có/không sử dụng schema trong Converter key
    • Converter value: chọn giá trị value cho converter
    • Converter value schema enable: chọn giá trị có/không sử dụng schema trong Converter value

image-5

Kafka Topic: Hiển thị danh sách các Kafka Topic mà CDC Service sẽ lắng nghe để nhận dữ liệu thay đổi từ Source Connector.

  • Kafka topic: Danh sách các topic được cấu hình để CDC Service subscribe.
    • Add (+): Thêm một Kafka Topic mới vào danh sách.
    • Delete (×): Xóa toàn bộ danh sách Kafka Topic đã cấu hình.
    • Topic: Tên Kafka Topic nhận dữ liệu thay đổi từ Source Connector.
    • Action: Xóa Kafka Topic tương ứng khỏi danh sách.
    • Filter: Tìm kiếm Kafka Topic theo tên.
    • Chú ý: Giới hạn chỉ lấy tối đa 100 topic

image-6

Bước 6: Nhấn Next để chuyển qua màn Additional Properties

Nhập các thông tin sau:

  • Timezone: chọn Timezone phù hợp của dữ liệu từ database nguồn

image-7

  • Type: chọn loại Database source

image-8

  • Task max: số task xử lý cùng nhau

image-9

  • Schema: tên Schema

image-10

  • Map table: Hiển thị danh sách mapping giữa Kafka Topic và Table trên cơ sở dữ liệu đích.
    • Topic: Tên Kafka Topic chứa dữ liệu thay đổi từ Source Connector.
    • Create new: Bật để tạo mới table trên cơ sở dữ liệu đích nếu table chưa tồn tại; tắt để sử dụng table hiện có.
    • Table: Tên table trên cơ sở dữ liệu đích dùng để đồng bộ dữ liệu từ Topic tương ứng.
    • Filter: Tìm kiếm theo tên Topic hoặc Table.
    • Chú ý: tích chọn Enable filter error để lọc những table name đang lỗi

image-11

  • Mode (required): Hành vi của Connector khi không thể xử lý được message
    • None: Connector sẽ bỏ qua các messages không thể sink vào CSDL
    • All: Các message lỗi sẽ được gửi vào topic được nhập

image-12

Bước 7. Nhấn Next để chuyển qua màn Review

image-13

Bước 8: Kiểm tra thông tin và nhấn nút Create để hoàn thành việc tạo connector.