ในการใช้งาน OpenMetadata ในกรณีที่ data source ของเราเป็น Database PostgreSQL เราจะต้องทำการตรวจสอบการติดตั้ง extension ที่มีชื่อว่า “pg_stat_statements” ก่อนเสมอก่อนที่จะทำการสร้าง service และ ingestion

pg_stat_statements เป็น PostgreSQL extension ที่ใช้มอนิเตอร์และตรวจสอบการใช้งาน Query ใน Database โดยหน้าที่จะเก็บ query ทั้งหมดที่ถูกเรียกใช้รวมถึง
บันทึกเวลาในการประมวลผลและทรัพยากรที่ใช้ของ query แต่ละตัว

โดยสามารถตรวจสอบได้จากการใช้คำสั่ง sql

SELECT * FROM pg_available_extensions WHERE name = 'pg_stat_statements';

หากพบว่ายังไม่มีการติดตั้งสามารถใช้คำสั่งต่อไปนี้ได้เลยเพื่อเพิ่มตัว pg_stat_statements ไปยัง database

CREATE EXTENSION IF NOT EXISTS pg_stat_statements;

และทำการตั้งค่า Privileges

GRANT SELECT ON pg_stat_statements TO public; #-- หรือแทน public ด้วย role/user

และในตอนสร้าง ingestion metadata หากไม่ต้องการเห็นข้อมูลในส่วนของ pg_stat_statements ที่อาจจะปะปนเข้ามากับตัว data source ที่จะใช้งานให้เราทำการ exclude ตัว table ที่เป็นของระบบซึ่งจะขึ้นต้นด้วย “pg_stat_statements”

หากทำการ ingestion แล้วพบ schema ที่ชื่อว่า information_schema ตัว information_schema เป็น system schema ทำหน้าที่เป็นแหล่งข้อมูล Metadata ของฐานข้อมูล โดยช่วยดึงข้อมูลเกี่ยวกับโครงสร้าง schema, table, column, และอื่น ๆ

หรือหากเรามีปัญหาหรือข้อสงสัยในการใช้งานสามารถ join เข้าไปยัง Slack ของ OpenMetadata เพื่อสอบถามการใช้งานได้เลยยย https://slack.open-metadata.org/

ผู้เขียนใช้งาน Openmetadata Version 2.5.13

Reference

• OpenMetadata: The Best Open Source Data Catalog Solution
• Postgres
• Extracting OpenMetadata Product Info

มาทำ Python Library ขึ้น PyPI แบบง่ายๆกันเถอะ

ในบทความนี้เราจะมาทำ CI/CD automate deploy สำหรับการนำ python library ขึ้นไปบน TestPyPI เพียงแค่เรา comit และ push source code ไปที่ github ก็จะทำการ publish ขึ้นไปที่ TestPyPI แบบอัตโนมัติ

สิ่งที่เราต้องเตรียมในรอบนี้ก็จะมี

1. Python code ที่เราต้องการจะ publish ขึ้นไปไว้ที่ PyPI
2. account ของ TestPyPI ที่มีการยืนยันตัวตนเรียบร้อยแล้ว
3. API Token ของ TestPyPI ที่เราได้จากรอบที่แล้ว
4. account ของ Github
5. Repository ใน Github ที่เอาไว้เก็บ Python code ของเรา

**หากเราจำ API Token ไม่ได้สามารถสร้างใหม่และกำหนด scope ให้ชี้มาที่ package ของเราได้

ในการทำ CI/CD ครั้งนี้เราจะใช้เครื่องมือ Github Actions ซึ่งเป็นแพลตฟอร์มการทำงานอัตโนมัติที่ช่วยให้เราสร้าง workflow ทำงานต่าง ๆ แบบโดยอัตโนมัติเมื่อมีการเปลี่ยนแปลงใน repository เช่น การรันทดสอบโค้ด, การ build, การ deploy, และการตรวจสอบโค้ด โดย GitHub Actions มีความยืดหยุ่นและสามารถปรับแต่งได้ตามต้องการผ่านการใช้ไฟล์ workflow ในรูปแบบ YAML ซึ่งถูกเก็บไว้ในโฟลเดอร์ .github/workflows/ ของ repository ที่เราสร้าง

เตรียม API Token

โดยเราจะทำการประกาศตัวแปรสำหรับเรียกใช้งาน API Token ของ TestPyPI ใน repository ของเราให้ไปที่เมนู Setting -> Secrets and variables จากนั้นเลือก new repository secret และนำ API Token ที่เรามีไปเก็บไว้ใน Secret

ตั้งค่า Permissions ให้ Workflow

ไปที่เมนู Settings -> Actions -> general

จากนั้นไปที่ Workflow permissions และทำการเลือก Read and write permissions จากนั้นทำการ save ได้เลย

จากนั้นไปที่เมนู Actions -> new workflow

หลังจากนั้นทำการ commit changes เพื่อเพิ่มไฟล์ workflow เข้าไปที่ repository ของเรา

ทีนี้เราจะปรับการทำงานของ workflow กันโดยจะให้มีการ trigger ทุกครั้งที่มีการ push code ขึ้นมาที่ branch “main” และทำการ deploy ขึ้นไปที่ TestPyPI

ไปที่ไฟล์ python-publish.yml ที่อยู่ใน <ชื่อ repository>/.github/workflows แล้วให้สังเกตุ code ตรง

on:
  release:
    types: [published]

ให้ทำการปรับใหม่เป็น

on:
  push:
    branches:
      - main

เท่านี้เราก็จะได้ trigger ที่ workflow จะเริ่มทำงานเมื่อมีการ push code เข้ามาที่ branch main แล้วจากนั้นทำการแก้ไขการเรียกใช้งาน API Token และปลายทางให้ชี้ไปที่ TestPypi ให้เราไปดู code ตรงส่วนนี้

with:
        user: __token__
        password: ${{ secrets.PYPI_API_TOKEN }}

ให้ทำการปรับเป็น

with:
        user: __token__
        password: ${{ secrets.TEST_PYPI_API_TOKEN }}
        repository_url: https://test.pypi.org/legacy/

โดยตรงตัวแปร password จะเป็นการเรียกใช้งานตัวแปร API Token ที่เราได้ประกาศไว้ที่ secret ก่อนหน้านี้ และเพิ่มตัวแปร repository_url เพื่อให้ชี้เป้าหมายไปยัง TestPyPI

ตอนนี้เราได้เตรียมการเรียบร้อยแล้วมาลองทดสอบการ push code กันเมื่อเราทำการแก้ไขและทำการ push ไปที่ branch main แล้วให้ไปดูที่เมนู Actions จะเห็นได้ว่า workflow ของเราถูกเรียกใช้งานแล้ว

ทีนี้เราเข้าไปตรวจสอบที่ TestPyPI และไปที่ project ของเราจะพบว่า v0.0.3 ขึ้นมาแล้วววววว

***ข้อสังเกตุ หากเรามีการแก้ไข code และ deploy ไปยัง TestPyPI เราต้องทำการแก้ไขเลข version ที่ไฟล์ pyproject.toml ด้วยหากไม่ปรับ code ที่เราแก้และ push ขึ้นไปจะไม่มีการ deploy ขึ้นไปยัง TestPyPI ตรงจุดนี้เราหากสามารถแยก branch ที่จะใช้ในการ deploy code และ publish ไปยัง TestPyPI ได้เพื่อไม่ให้ส่งผลต่อการแก้ไขและพัฒนา code

เท่านี้เราก็สามารถ Automate Deploy ผ่านการ push ไปยัง GitHub ด้วยการใช้เครื่องมืออย่าง GitHub Actions สามารถลดข้อผิดพลาดจากการ deploy แบบ manual และทำให้การพัฒนาซอฟต์แวร์ของเรามีความต่อเนื่อง (CI/CD) โดยไม่สะดุดและทำให้สามารถเผยแพร่ library ให้กับผู้ใช้ได้ทันทีโดยไม่ต้องคอยจัดการด้วยตนเอง ช่วยให้เราสามารถมุ่งเน้นไปที่การพัฒนาและปรับปรุงคุณภาพ code ได้มากขึ้น

อ้างอิง GitHub Actions document

*** ข้อระวังการใช้งาน GitHub Actions จะมีการจำกัดเวลาไว้ที่ 2000นาทีต่อเดือน

PyPI (Python Package Index) คือ repository หรือแหล่งเก็บแพ็กเกจของ Python ที่ใช้สำหรับการจัดเก็บและเผยแพร่ไลบรารี Python ให้ผู้ใช้งานอื่นๆ สามารถดาวน์โหลดและติดตั้งผ่านเครื่องมืออย่าง pip

โดยในบทความนี้เราจะทำการ publish ขึ้นไปไว้ที่ TestPyPI ซึ่งจะเป็นอีก instance หนึ่งแยกกับ PyPI ทำให้เราสามารถทดสอบการเผยแพร่ได้โดยไม่ส่งผลกระทบต่อ PyPI หลัก

สิ่งที่เราต้องเตรียมไว้ก่อนเลยก็จะมี

1. Python code ที่เราต้องการจะ publish ขึ้นไปไว้ที่ PyPI
2. account ของ TestPyPI ที่มีการยืนยันตัวตนเรียบร้อยแล้ว
3. เครื่องมือที่จำเป็น pip

สร้างโครงสร้างของ directory Python โดยจะมีหน้าตาแบบนี้

ทีนี้เราจะมาดูไฟล์ที่สำคัญและจำเป็นต้องมีแต่ละส่วนในโครงสร้าง package ทำหน้าที่อะไรบ้าง:

• __init__.py: บอกว่าโฟลเดอร์เป็น Python package
• README.md: เอกสารอธิบายข้อมูลเกี่ยวกับแพ็กเกจ
• LICENSE: ใบอนุญาตที่ใช้สำหรับ library ของเรา
• pyproject.toml: ใช้สำหรับการตั้งค่า build system
• tests/ : ใช้สำหรับเก็บไฟล์ test ตอนนี้เราปล่อยว่างไว้ก่อน
• example.py : เป็นไฟล์ python ที่เราเขียนไว้

เลือกเครื่องมือสำหรับ build backend

โดยตามเอกสาร official จะมีการระบุเครื่องมือไว้ทั้งหมด4ตัวได้แก่ Hatchling, Setuptools , Flit และ PDM โดยแต่ละตัวมีจุดประสงค์คล้ายกัน แต่มีความแตกต่างกันในด้าน ความยืดหยุ่น และวิธีการใช้งาน ซึ่งตัวที่เราเลือกใช้งานคือตัว Hatchling ที่มีการใช้งานง่าย และเหมาะกับ project เล็กที่ไม่ซับซ้อนมากโดยจะมีการเรียกใช้แบบนี้

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

โดยจะนำไปใส่ในไฟล์ pyproject.tomlที่จะอธิบายในขั้นตอนต่อไป

การตั้งค่าไฟล์ pyproject.toml

จากนั้นเราจะทำการแก้ไขไฟล์ pyproject.toml โดยจะประกอบไปด้วย

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

[project]
name = "ชื่อ package ของเรา"
version = "0.0.1"
authors = [
  { name="ชื่อเจ้าของ", email="author@example.com" },
]
description = "อธิบายรายละเอียดของ package เรา"
readme = "README.md"
requires-python = ">=3.8"
classifiers = [
    "Programming Language :: Python :: 3",
    "License :: OSI Approved :: MIT License",
    "Operating System :: OS Independent",
]

[project.urls]
Homepage = "https://github.com/pypa/sampleproject"
Issues = "https://github.com/pypa/sampleproject/issues"

• name ชื่อ package ของเรา ซึ่งอาจเป็นชื่อใดก็ได้ แต่ต้องไม่มีชื่อแพ็กเกจนี้อยู่ใน PyPI อยู่แล้ว
• version เป็นการกำหนด version ของ package เรา หรืออาจจะมีการกำหนดมาจากส่วนอื่นในกรณีที่เรามีการใช้ git
• authors ข้อมูลเจ้าของ package โดยจะมีระบุชื่อกับ email ของเราลงไป
• description อธิบายข้อมูล package ของเรา
• readme เป็นค่า path ของไฟล์ README.md ที่เป็นไฟล์เก็บข้อมูลรายละเอียดเกี่ยวกับ package ของเรา
• requires-python ความต้องการ version ขั้นต่ำของ python ที่ต้องการในการติดตั้ง package ของเรา
• classifiers เป็นการจัดหมวดหมู่ของ package เช่น สถานะของ package , ระบบปฏิบัติการที่รองรับ, เวอร์ชัน Python ที่รองรับ, ประเภทของซอฟต์แวร์ โดยสามารถดูรายการหมวดหมู่ได้จากเอกสารนี้ List of classifiers
• urls รายการ link url เพิ่มเติมเกี่ยวกับ เอกสารประกอบ ที่มาสำหรับแสดงบน PyPI

สร้าง README.md

ระบุข้อมูลรายละเอียดของ package เรา

# Example Package

This is a simple example package.

กำหนด LICENSE

จากนั้นเราก็ต้องทำการกำหนด LICENSE เพื่อเป็นการบอกว่า library ของเรานั้นอนุญาตให้ให้ใช้ library ของเราทำอะไรได้บ้างและใช้โดยใคร ซึ่งตัว license เราสามารถไปดูข้อมูลเพิ่มเติมได้จาก choosealicense

การ build package

เมื่อเราได้กำหนดค่าไฟล์ต่างๆเรียบร้อย รวมถึงไฟล์ python ของเราเองพร้อมให้คนอื่นนำไปใช้งานแล้วในขั้นตอนต่อไปจะเป็น ทำการติดตั้งเครื่องมือที่ใช้สำหรับการ build Python packages โดยใช้คำสั่งตามนี้ได้เลย

py -m pip install --upgrade build

ทำการ run คำสั่ง build โดยก่อน run ให้ตรวจสอบที่อยู่ directory ก่อนจะ run ว่าต้องอยู่ใน directory เดียวกับไฟล์ pyproject.toml ด้วย

py -m build

หาก build สำเร็จเราจะได้ไฟล์มาอยู่ใน directory dist/ ซึ่งจะมี 2 ตัวได้แก่ไฟล์นามสกุล .whl และ .tar.gz

.whl (wheel) เป็นไฟล์ที่ถูก build และพร้อมใช้งานแล้ว ซึ่งทำให้ pip สามารถติดตั้ง package ได้เลยโดยไม่ต้อง build package ขึ้นใหม่

.tar.gz เป็นไฟล์ที่ประกอบไปด้วย source code ของ package และจะต้องถูก build ขึ้นใหม่เมื่อมีการติดตั้งโดย pip

สร้าง API Token

เมื่อเรามี account ของ TestPyPI ที่มีการยืนยันตัวตนเรียบร้อยแล้วให้ทำการสร้าง API Token ผ่านเมนู Account Setting โดยทำการกำหนดชื่อ project และ scope การใช้งาน

เริ่ม Upload Package

ก่อนอื่นเราทำการติดตั้ง twine สำหรับใช้ในการ upload package ของเราขึ้นไปยัง TestPyPI

py -m pip install --upgrade twine

Twine เป็นเครื่องมือที่ใช้สำหรับอัปโหลด Python packages ขึ้นไปยัง PyPI หรือ TestPyPI ได้อย่างปลอดภัย โดยผ่าน HTTPS

หลังจากติดตั้ง twine เรียบร้อยแล้วเราจะมาทำการ upload package ของเราขึ้นไปยัง TestPyPI ก่อนอื่นเลยให้เราทำการตรวจสอบไฟล์ package โดยการใช้คำสั่ง

python3 -m twine check dist/*

หากขึ้น passed แปลว่าพร้อม upload แล้วให้ทำการ upload ได้เลย

py -m twine upload --repository testpypi dist/*

จากคำสั่งที่ run ไปจะเป็นการนำไฟล์ที่ผ่านการ build แล้วใน folder dist/ upload ขึ้นไปที่ repository testpypi และจะมีการถามหาตัว API Token ที่เราได้ทำการสร้างไว้ก่อนหน้านี้ให้ทำการ copy แล้ว past ไปที่ terminal ได้เลย

เมื่อเรา upload สำเร็จแล้วที่เมนู your projects จะปรากฏ package ที่เราได้ทำการ upload ไปดังภาพ

ตอนนี้ Library ของเราสามารถใช้งานผ่าน pip ได้แล้ว โดยสามารถ copy คำสั่ง pip install แล้วลอง import ใช้งานกันได้เลย

หากต้องการ publish ขึ้น PyPI ปกติสามารถใช้วิธีตามนี้ได้เพียงแค่เปลี่ยน repository จาก TestPyPI ไปเป็น PyPI โดย account ของ PyPI ปกติจะแยกกับ TestPyPI เราต้องไปสมัครและยืนยันตัวตนใหม่อีกรอบ

py -m twine upload dist/*
รวมถึงตอนใช้งาน twine upload ไม่จำเป็นต้องใช้ --repository 
ต่อท้าย เนื่องจากมีค่า default เป็น pypi อยู่แล้ว

การสร้างและเผยแพร่ Python library ขึ้นไปยัง PyPI ไม่ใช่เรื่องที่ยากอย่างที่คิด หากเราทำตามขั้นตอนที่ถูกต้อง ไม่ว่าจะเป็นการเตรียมโครงสร้าง package การเขียนไฟล์ตั้งค่า และการอัปโหลดไปยัง PyPI โดยเครื่องมืออย่าง twine จะช่วยให้เราสามารถทำทุกขั้นตอนได้อย่างเป็นระบบ

การเผยแพร่ library ของเราบน PyPI ไม่เพียงแค่ช่วยให้เราสามารถแบ่งปันโค้ดให้กับคนอื่นๆ ได้ใช้งาน แต่ยังช่วยให้library ของเราได้มีการปรับปรุงและพัฒนา จากการช่วยกันของชุมชนผู้ใช้งาน นอกจากนี้ การมีเอกสารรายละเอียดการใช้งานที่ชัดเจนและการปรับปรุงอย่างสม่ำเสมอจะช่วยให้ library ของเรามีคุณภาพและได้รับความนิยมใช้งานมากขึ้น

ดังนั้น หากเรามี library หรือ code ที่น่าสนใจ อย่ารอช้าลองนำขึ้นไปที่ PyPI แล้วแบ่งปันให้กับชุมชนผู้ใช้งาน Python ได้ใช้งานกันเถอะ!

อ้างอิง Python Packaging User Guide

• เครื่องที่ติดตั้ง CentOS7 ไว้
• Cloudera account with licence(ถ้ามีหากไม่มีจะใช้เป็นแบบ trial แทน)
• Java open jdk
• python
• Database for config (PostgreSQL)+user to access

ก่อนจะทำการติดตั้งให้ทำการตรวจสอบ version ของเครื่องมือต่างๆที่จะใช้ได้ใน https://supportmatrix.cloudera.com/

Step 1. Config repository for cloudera manager ( version. 7.9.5 )

เข้าไปยัง directory /etc/yum.repos.d/ แล้วใช้คำสั่ง

sudo wget https://[username]:[password]@archive.cloudera.com/p/cm7/[**Cloudera Manager version**]/redhat8/yum/cloudera-manager.repo

หลังจากนั้นทำการ Import the repository signing GPG key โดยใช้คำสั่ง

sudo rpm --import https://[username]:[password]@archive.cloudera.com/p/cm7/[**Cloudera Manager version**]/redhat7/yum/RPM-GPG-KEY-cloudera

Step 2. ติดตั้ง java

ทำการติดตั้ง java โดยใช้คำสั่ง

yum install java-1.8.0-openjdk-devel

Step 3. ติดตั้ง Cloudera Manager Server

ติดตั้ง Cloudera manager server โดยใช้คำสั่ง

yum install cloudera-manager-daemons cloudera-manager-agent cloudera-manager-server

Step 4. ติดตั้ง Python สำหรับ hue

ทำการติดตั้ง package

• openssl-devel
• bzip2-devel
• libffi-devel
• zlib-devel

yum install gcc openssl-devel bzip2-devel libffi-devel zlib-devel -y

หลังจาก install package เสร็จเรียบร้อยแล้วจะทำการติดตั้ง python ต่อโดยให้ย้าย directory ไปที่ opt โดยใช้คำสั่ง

cd /opt

จากนั้นทำการ download python โดยใช้คำสั่ง

curl -O https://www.python.org/ftp/python/3.8.12/Python-3.8.12.tgz

แล้วทำการแตกไฟล์ด้วยคำสั่ง

tar -zxvf Python-3.8.12.tgz

หลังจากแตกไฟล์เรียบร้อยให้เข้าไปใน folder โดยใช้คำสั่ง

cd Python-3.8.12

จากนั้นใช้คำสั่งดังต่อไปนี้เพื่อติดตั้ง python

./configure --enable-shared

หลังจากติดตั้งเรียบร้อยแล้วทำการ compile file ต่อโดยใช้คำสั่ง

make

จากนั้นใช้คำสั่ง

make install

คัดลอกไฟล์ libpython3.8.so ไปไว้ใน /lib64/ โดยใช้คำสั่ง

cp --no-clobber ./libpython3.8.so* /lib64/

ทำการปรับ permission file โดยใช้คำสั่ง

chmod 755 /lib64/libpython3.8.so*

และ

chmod -R 755 /usr/local/lib/python3.8

Step 5. ติดตั้ง Database สำหรับ config

โดยจะรองรับ PostgreSQL , MySQL , MariaDB , Oracle Database โดยในการติดตั้งครั้งนี้จะเลือกใช้ PostgreSQL สำหรับใช้งานกับ cloudera

เริ่มจากทำการติดตั้ง postgresql-server โดยใช้คำสั่ง

sudo yum install postgresql-server

จากนั้นทำการติดตั้ง Postgres JDBC Driver โดยทำการตรวจสอบ version ของ driver ได้จาก https://jdbc.postgresql.org/ และใช้คำสั่ง

yum install postgresql-jdbc<compatible_version>

ทำการ update yum อีกรอบ

yum update

และทำการเปลี่ยนชื่อไฟล์ driver และย้ายตำแหน่งไฟล์โดยใช้คำสั่ง

cp /usr/share/java/postgresql-jdbc.jar /usr/share/java/postgresql-connector-java.jar

ตรวจสอบไฟล์ .jar โดยใช้คำสั่ง

ls /usr/share/java/postgresql-connector-java.jar 

ปรับ permission ของไฟล์ .jar โดยใช้คำสั่ง

chmod 644 /usr/share/java/postgresql-connector-java.jar

จากนั้นจะทำการ install psycopg2 Python package สำหรับ PostgreSQL-backed Hue

โดยให้ไปที่ directory ที่มีการติดตั้ง cloudera-manager.repo ไว้ ( ซึ่งในการลงครั้งนี้ไฟล์จะอยู่ที่ /etc/yum.repo.d ) จากนั้นทำการแก้ไขไฟล์โดยใช้คำสั่ง

nano cloudera-manager.repo

โดยในไฟล์ config ของ cloudera-manager.repo จะมีการเพิ่มค่าไปดังนี้

[postgresql10]
name=Postgresql 10
baseurl=https://archive.cloudera.com/postgresql10/redhat8/
gpgkey=https://archive.cloudera.com/postgresql10/redhat8/RPM-GPG-KEY-PGDG-10
enabled=0 
gpgcheck=1
module_hotfixes=true 

ทำการ restart ตัว database

sudo systemctl restart postgresql-14

สร้าง database config สำหรับตัว cloudera

ทำการเข้าใช้งาน posgreSQL ด้วยคำสั่ง

sudo -u postgres psql

สร้าง user สำหรับใช้งานใน database โดยใช้คำสั่ง

CREATE ROLE <user> LOGIN PASSWORD '<password>';



------------------- ตัวอย่าง ----------------------
1.============== Cloudera Manager Server =====
CREATE ROLE scm LOGIN PASSWORD '<password>';

2.============== Reports Manager ====================
CREATE ROLE rman LOGIN PASSWORD '<password>';

3.============== Ranger RHEL/CentOS/Ubuntu ====================
CREATE ROLE rangeradmin LOGIN PASSWORD '<password>';

4.============== Ranger KMS RHEL/CentOS ====================
CREATE ROLE rangerkms LOGIN PASSWORD '<password>';

5.============== Hue ====================
CREATE ROLE hue LOGIN PASSWORD '<password>';

6.============== Hive Metastore Server ====================
CREATE ROLE hive LOGIN PASSWORD '<password>';

7.============== Oozie ====================
CREATE ROLE oozie LOGIN PASSWORD '<password>';

8.============== Data Analytics Studio (DAS) Supported with PostgreSQL only. ====================
CREATE ROLE das LOGIN PASSWORD '<password>';

9.============== Schema Registry ====================
CREATE ROLE schemaregistry LOGIN PASSWORD '<password>';

10.============== Streams Messaging Manager ====================
CREATE ROLE smm LOGIN PASSWORD '<password>';

ทำการสร้าง database ขึ้นมาโดยใช้คำสั่ง

CREATE DATABASE <database> OWNER <user> ENCODING 'UTF8';

------------------- ตัวอย่าง ----------------------
1.============== Cloudera Manager Server ====================
CREATE DATABASE scm OWNER scm ENCODING 'UTF8';

2.============== Reports Manager ====================
CREATE DATABASE rman OWNER rman ENCODING 'UTF8';

3.============== Ranger RHEL/CentOS/Ubuntu ====================
CREATE DATABASE ranger OWNER rangeradmin ENCODING 'UTF8';

4.============== Ranger KMS RHEL/CentOS ====================
CREATE DATABASE ranger OWNER rangerkms ENCODING 'UTF8';

5.============== Hue ====================
CREATE DATABASE hue OWNER hue ENCODING 'UTF8';

6.============== Hive Metastore Server ====================
CREATE DATABASE hive OWNER hive ENCODING 'UTF8';

7.============== Oozie ====================
CREATE DATABASE oozie OWNER oozie ENCODING 'UTF8';

8.============== Data Analytics Studio (DAS) Supported with PostgreSQL only. ====================
CREATE DATABASE das OWNER das ENCODING 'UTF8';

9.============== Schema Registry ====================
CREATE DATABASE schemaregistry OWNER schemaregistry ENCODING 'UTF8';

10.============== Streams Messaging Manager ====================
CREATE DATABASE smm OWNER smm ENCODING 'UTF8';

สร้างทีละ database ตามตารางด้านล่าง

หากมีการใช้งาน PostgreSQL 8.4 หรือใหม่กว่าให้ทำการตั้งค่าตัวแปรของ Hive Metastore และ Oozie ที่ชื่อว่า standard_conforming_strings เป็น off โดยใช้คำสั่ง

ALTER DATABASE <database> SET standard_conforming_strings=off;

========== ตัวอย่าง ==============
ALTER DATABASE hive SET standard_conforming_strings=off;
ALTER DATABASE Oozie SET standard_conforming_strings=off;

Step 6. ตั้งค่า config สำหรับ cloudera manager database

ในการตั้งค่า config จะมีไฟล์ script สำหรับ run ที่ชื่อว่า scm_prepare_database.sh โดยการเรียกใช้งานจะใช้คำสั่งดังต่อไปนี้

sudo /opt/cloudera/cm/schema/scm_prepare_database.sh [options] <databaseType> <databaseName> <databaseUser> <password>

โดยค่า parameter จะประกอบไปด้วย

ตัวอย่างคำสั่งสำหรับการตั้งค่า config ของ scm service ไปยัง database

==== กรณีที่ cloudera manager กับ database config อยู่ภายในเครื่องเดียวกัน ====
sudo /opt/cloudera/cm/schema/scm_prepare_database.sh mysql scm scm

==== กรณีที่ cloudera manager กับ database config อยู่คนละเครื่อง ====
sudo /opt/cloudera/cm/schema/scm_prepare_database.sh mysql -h <ip/host name เครื่อง database> - scm-host <hostname เครื่อง cloudera> scm scm


============== Cloudera Manager Server ====================
sudo /opt/cloudera/cm/schema/scm_prepare_database.sh postgresql -h <dbhost> -P 5432 --scm-host <clouderamanager host> scm scm

Step 7. start cloudera scm server

ก่อนอื่นทำการตรวจสอบ service cloudera-scm-server โดยใช้คำสั่ง

sudo systemctl status cloudera-scm-server

จากนั้นทำการ start ตัว server ขึ้นมา

sudo systemctl start cloudera-scm-server

และตั้งค่าให้ตัว server ทำการ start แบบ automatic เมื่อมีการ reboot เครื่องโดยใช้คำสั่ง

sudo systemctl enable cloudera-scm-server

ทำการตรวจสอบ status server อีกครั้ง

sudo systemctl status cloudera-scm-server

หลังจาก start server เรียบร้อยแล้วสามารถเข้าใช้งานได้ผ่าน url ตามนี้

http://<server_host>:7180

========Default User & Password สำหรับ login =====
Username: admin
Password: admin

**โดยในการเข้าใช้งานครั้งแรกจะมีหน้าต่างขึ้นมาให้เลือกในส่วนของการใช้งาน แบบมีlicence กับ แบบ trial

**หากต้องการดู log ของตัว server สามารใช้คำสั่ง

sudo tail -f /var/log/cloudera-scm-server/cloudera-scm-server.log

เพื่อคอยตรวจสอบ error หรือการแจ้งเตือนอาการผิดปกติได้

Apache NiFi สามารถใช้งานได้ในกรณีการประมวลผลข้อมูลแบบครบวงจร (end-to-end data flow) ซึ่งรวมถึงการเชื่อมต่อและการควบคุมกระบวนการข้อมูล โดยสามารถทำการติดตามสถานะ การควบคุม การจัดเก็บ การกระจายและการประมวลผลข้อมูลได้อย่างง่ายดาย นอกจากนี้ยังสามารถทำงานแบบเรียลไทม์ที่ดีกับการควบคุม การตรวจจับข้อผิดพลาด และการกำหนดข้อกำหนดเพื่อให้ได้ความสามารถในการประมวลผลแบบอัตโนมัติได้

Install in centos7

โดยในการติดตั้ง Apache NiFi เราจะต้องมีการติดตั้งตัว java พร้อมกับประกาศ JAVA_HOME ให้เรียบร้อยจากนั้นทำการ download ตัว Apache NiFi จาก https://archive.apache.org/dist/nifi/ โดยเลือก version ที่ต้องการติดตั้ง และทำการ download ตัวที่เป็น bin

โดยจะยกตัวอย่างเป็นการใช้ version 1.21.0 ก็จะ download ไฟล์ตามนี้ nifi-1.21.0-bin.zip

ทำการตรวจสอบว่าภายในเครื่องเรามีการประกาศ JAVA_HOME ไว้เรียบร้อยแล้วโดยใช้คำสั่ง

echo $JAVA_HOME

จากนั้นทำการแตกไฟล์ zip ที่download มาโดยใช้คำสั่ง

unzip nifi-1.21.0-bin.zip

หลังจากที่ทำการแตกไฟล์เรียบร้อยแล้วเราจะได้ folder เพิ่มขึ้นมาโดยให้เราเข้าไปที่ directory bin เพื่อทดสอบ start ตัว Apache NiFi โดยใช้คำสั่ง

./nifi.sh start

หาก run สำเร็จเราสามารถเข้าในงาน Apache NiFi ได้ผ่าน web browser url

https://localhost:8443/nifi/

โดยจะมีค่า port default เป็น 8443

ในการ login เข้าใช้งานหากเราไม่ได้มีการ set user จะต้องใช้ username และ password จาก log ซึ่งเราสามารถเรียกดู generate username/password สำหรับloginได้จากไฟล์ log nifi-app.log ที่อยู่ใน directory logs จากนั้นใช้คำสั่ง

cat nifi-app.log | grep Generated

ดังนั้นหากต้องการกำหนด username password สำหรับใช้งานในตอนที่ start ให้เราใช้คำสั่ง

nifi.sh set-single-user-credentials <username กำหนดเอง> <password กำหนดเอง>

แทนจากเดิมที่เป็นแค่ start ธรรมดา

• Docker Desktop(Download)
• Debezium โดยเลือก version ที่ต้องการและ download ไฟล์ debezium-connector-mysql-<version>.Final-plugin.zip
• **หากต้องการทำ sink connector ที่เป็น mysql ต้องใช้ JDBC Driver สำหรับใช้เชื่อมต่อ database ใช้ mysql-connector-java ที่เป็นไฟล์ .jar / .pom (download) แต่ภายในบทความจะใช้งาน postgresql เป็น database สำหรับ sink

Cloudera CSP คืออะไร

Cloudera CSP (Cloudera Streaming Platform) เป็นแพลตฟอร์มการสตรีมข้อมูลแบบองค์รวม (end-to-end streaming platform) ที่ออกแบบมาเพื่อให้ผู้ใช้สามารถสร้างและบริหารจัดการระบบการสตรีมข้อมูลได้อย่างมีประสิทธิภาพและง่ายดาย โดยใช้เทคโนโลยี Apache Flink และ Apache NiFi เป็นหลัก

Step1. ติดตั้ง CSP Community Edition

หลังจากที่เราได้ทำการติดตั้ง docker desktop เรียบร้อยแล้ว จะทำการเตรียมไฟล์ docker-compose.yml ของ CSP โดยสามารถ download ได้จาก Downloads Page of CSP Community Edition. จากนั้นสร้าง folder ชื่อ cspce และนำไฟล์ docker-compose.yml ไปไว้ด้านในหลังจากนั้นทำการเปิด terminal หรือ cmd และเข้าไปที่ folder cspce และทำการ run คำสั่ง

docker-compose up -d

เมื่อติดตั้งเรียบร้อยจะได้ docker container แบบเต็มระบบ โดบหลังจากนี้เราสามารถใช้งานตัว User Interface ผ่าน url ดังต่อไปนี้

Schema Registry :            http://localhost:7788
Streams Messaging Manager :  http://localhost:9991
Streaming SQL Builder:       http://localhost:18121
                             Username: admin
                             Password: admin

Step2. Update debezium version in docker

ทำการแตกไฟล์ debezium-connector-mysql-<version>.Final-plugin.zip

จากนั้นที่ docker desktop ให้เราเข้าไปที่เครื่อง kafka-connect-1 แล้วไปที่แท็บ file

และเข้าไปยัง opt/connect/plugin/libs จะพบกับ folder ที่ติดตั้ง debezium ไว้ชื่อว่า debezium-connector-mysql ให้ทำการ delete ทิ้งไปและทำการ import ไฟล์ที่ได้จากการแตกไฟล์เข้าไปวางแทนที่ชื่อเดิมและทำการ restart เครื่อง

Step3. create connector and connect to source

หลังจากทำการอัพเดท debezium version เรียบร้อยแล้วทำการเข้ามายังหน้า ui ของ Streams Messaging Manager ผ่าน

http://localhost:9991

และเข้าไปยัง menu connector ไปที่ new connector

แล้วทำการเลือก source template ที่ต้องการสร้าง

ระบุการตั้งค่า source database ให้ครบถ้วน

หลังจากระบุครบถ้วนแล้วให้ทำการ Validate config ด้วยปุ่ม validate ด้านขวาบนและทำการ next จากนั้นตรวจสอบอีกครั้งจึงทำการ deploy

Step4.(optional) create connector for sink

การสร้าง sink connector จะเหมือนกับการสร้าง source connector โดยในครั้งนี้จะเลือกเป็น เชื่อมต่อไปยัง postgres database

จากนั้นใช้ import connector configuration

ทำการตรวจสอบค่าตัวแปรต่างๆแล้วทำการ validate ก่อนกด next และ deploy

Archive logs คือการสร้างสำเนาของ Redo logs และเก็บไว้ในที่เก็บข้อมูลที่แตกต่างจากที่เก็บ Redo logs ปกติ โดยทำการ Archive logs เป็นวิธีการทำ Backup and Recovery ที่มักจะถูกนำมาใช้งานในการกู้ข้อมูลที่สูญหายหรือเสียหาย

Redo logs คืออะไร?

Redo logs เป็นไฟล์ที่ใช้บันทึกข้อมูลการทำงาน/การกระทำ (transactional data) ที่เกิดขึ้นในฐานข้อมูล โดย Redo logs จะถูกสร้างขึ้นเมื่อมีการกระทำเกี่ยวกับข้อมูลในฐานข้อมูล เช่น เพิ่ม ลบ หรือแก้ไขข้อมูลในตาราง

Redo logs จะมีการเขียนลงไฟล์แบบ Circular ซึ่งเป็นการเขียนทับข้อมูลเก่าที่มีอยู่ในไฟล์ Redo logs แล้วเมื่อครบรอบการบันทึก

การใช้ Redo log นั้นมีประโยชน์อย่างมาก เช่น การ Backup and Recovery และช่วยลดความเสี่ยงที่จะสูญเสียข้อมูลในฐานข้อมูลได้

Supplemental Logging คืออะไร

เป็นการเพิ่มข้อมูลเพิ่มเติมเกี่ยวกับการเปลี่ยนแปลงข้อมูลในตาราง โดยจะเป็นการเพิ่ม key เข้าไปซึ่งจะช่วยให้ข้อมูลที่ได้จาก redo log สามารถนำไปใช้งานต่อได้ง่ายขึ้น

*การเปิดใช้งาน Archive logs และ Redo logs จะทำผ่าน SQL*Plus

Step 1. Archive logs

ตรวจสอบ LOG_MODE โดยใชเคำสั่ง

SELECT LOG_MODE FROM V$DATABASE

หากไม่มีการ setup ไว้ LOG_MODE จะมีค่าเป็น NOARCHIVELOG

ทีนี้จะเริ่มทำการกำหนด parameter สำหรับ archive logs โดยใช้คำสั่ง

ALTER SYSTEM SET db_recovery_file_dest_size = 10G;
ALTER SYSTEM SET db_recovery_file_dest = '/opt/oracle/oradata/ORCLCDB' scope=spfile;

db_recovery_file_dest_size -> จะเป็นการกำหนดขนาดของความจุที่จะใช้เก็บตัว archive logs

db_recovery_file_dest -> จะเป็นตัวกำหนด path directory สำหรับเก็บตัว archive logs

หลังจากกำหนด parameter แล้วเราจะทำการ restart database โดยใช้คำสั่ง

SHUTDOWN IMMEDIATE
STARTUP MOUNT

หลังจาก start up เรียบร้อยจะทำการเปิดใช้งานตัว archive logs โดยใช้คำสั่ง

ALTER DATABASE ARCHIVELOG;
ALTER DATABASE OPEN;
ARCHIVE LOG LIST;

Step2. Redo logs

หลังจากทำการเปิดการใช้งาน archive log แล้วจะทำการตั้งค่าการใช้งาน redo logs โดยใช้คำสั่ง

SELECT GROUP#, BYTES/1024/1024 SIZE_MB, STATUS FROM V$LOG ORDER BY 1;

โดยจะมีการส่งค่า log group กลับมาซึ่งจะมีการระบุขนาดและ status ของแต่ละ group ไว้และใช้คำสั่งต่อไปนี้เพื่อหาชื่อไฟล์ และที่อยู่ของไฟล์ log ว่าถูกเก็บไว้ใน directory ไหนในเครื่อง

SELECT GROUP#, MEMBER FROM V$LOGFILE ORDER BY 1, 2;

หลังจากนั้นเราจะทำการปรับขนาดของไฟล์ log จากเดิมที่มี default เป็น 200mb ให้เป็น 400mb โดยใช้คำสั่ง

ALTER DATABASE CLEAR LOGFILE GROUP 1;
ALTER DATABASE DROP LOGFILE GROUP 1;
ALTER DATABASE ADD LOGFILE GROUP 1 ('/opt/oracle/oradata/ORCLCDB/redo01.log') size 400M REUSE;
ALTER SYSTEM SWITCH LOGFILE;

หลังจากปรับเสร็จแล้วทำการตรวจสอบโดยใช้คำสั่ง

SELECT GROUP#, BYTES/1024/1024 SIZE_MB, STATUS FROM V$LOG ORDER BY 1;

Step3. Supplemental Logging

ปกติ Supplemental Log จะไม่ได้ถูกเปิดใช้งานดังนั้น ในการใช้งานร่วมกับ Debezium ต้องเปิดใช้งาน Supplemental Logging ก่อนจึงจะสามรถใช้งานได้โดยใช้คำสั่ง

ALTER DATABASE ADD SUPPLEMENTAL LOG DATA;

เตรียม os user สำหรับลง database โดยในที่นี้จะใช้เป็น user oracle

เตรียมพร้อมตัว install โดยในขั้นตอนแรกๆแนะนำให้ใช้งาน root user โดยใช้คำสั่ง

sudo su

yum -y update

เตรียม file preinstall และ install

Download ตัว rpm โดยใช้ curl

curl -o oracle-database-preinstall-19c-1.0-1.el7.x86_64.rpm https://yum.oracle.com/repo/OracleLinux/OL7/latest/x86_64/getPackage/oracle-database-preinstall-19c-1.0-1.el7.x86_64.rpm

ทำการ loacal install

yum -y localinstall oracle-database-preinstall-19c-1.0-1.el7.x86_64.rpm

Step 2. Install base rpm

Download file install ที่เป็น rpm จาก https://www.oracle.com/database/technologies/oracle-database-software-downloads.html
โดย ใช้ rpm ที่เป็น 19.3 — Enterprise Edition Linux x86–64

ทำการ install โดยใช้

yum -y localinstall oracle-database-ee-19c-1.0-1.x86_64.rpm

Step 3. Creating and Configuring

หลังจาก install เรียบร้อยจะทำการเริ่ม config database โดยใช้คำสั่ง

/etc/init.d/oracledb_ORCLCDB-19c configure

Step 4. Firewall

ทำการเพิ่ม port 1521 เข้าไป

firewall-cmd --add-port=1521/tcp --permanent

จากนั้นทำการ reload

firewall-cmd --reload

ตรวจสอบโดยใช้คำสั่ง

firewall-cmd --list-ports 1521/tcp

Step5.Edit Environment variables

ในขั้นตอนนี้จะเป็นการตั้งค่า ORACLE_HOME , PATH ,ORACLE_SID , LANG โดยจะต้องทำการสลับ user โดยใช้คำสั่ง

su oracle

ทำการแก้ไขไฟล์ .bashrc

nano ~/.bashrc

เพิ่มข้อความต่อไปนี้เข้าไปในไฟล์

export LANG=en_US.UTF-8
export ORACLE_HOME=/opt/oracle/product/19c/dbhome_1 
export PATH=$PATH:/opt/oracle/product/19c/dbhome_1/bin 
export ORACLE_SID=ORCLCDB

หลังจาก save แล้วทำการ refresh ด้วยคำสั่ง

source ~/.bashrc

Step 6. Listener

จากนั้นเราจะทำการตรวจสอบตัว listener ว่าทำงานปกติหรือไม่ด้วยคำสั่ง

lsnrctl status

หาก listener ยังไม่ถูกเปิดจะมีข้อความหน้าตาแบบนี้

ให้เราทำการ start โดยใช้คำสั่ง

lsnrctl start

Step 7. ทดลองใช้ sqlplus

ทำลองเข้าใช้งาน database ด้วยคำสั่ง

sqlplus / as sysdba

ทำการกำหนด password ให้กับ user system ด้วยคำสั่ง

alter user system identified by <กำหนด password>;

หลังจากนั่นสามาถทำการทดสอบ remote login ด้วย dbeaver ได้

Step 8. Create user/schema

ในส่วนนี้จะเป็นการสร้าง user สำหรับ remote เข้ามาใสช้งานกรณีที่ไม่อยากใช้ system โดยใน sql plus จะใช้คำสั่ง

CREATE USER <username> IDENTIFIED BY <password>;

เมื่อ success แล้วจะ grant ให้ user สามารถเข้าถึง Oracle Database ได้จาก remote ด้วยคำสั่ง

GRANT CREATE SESSION TO <username>;

ถ้าต้องการให้ user สามารถสร้าง table space และ objects อื่น ๆ ใน Oracle Database ได้ ให้ใช้คำสั่ง

GRANT ALL PRIVILEGES TO <username>;

การเชื่อมต่อ kafka กับ databaseภายนอกเราจะใช้ตัว connector ในการเชื่อมโดยตัว connector จะมีทั้งหมด3แบบได้แก่ connect-standalone , connect-mirror และ connect-distributed ใช้สำหรับการนำเข้า/ส่งออกข้อมูลจาก Kafka ไปยังระบบภายนอก

• connect-standalone : เป็นโหมดการทำงานของ Kafka Connect ที่ทำงานเป็นแบบ standalone หรือเดี่ยว ไม่มีการใช้ worker หลายๆตัวเหมือนกับ connect-distributed โดยการทำงานของมันจะเป็นการตั้งค่า worker เพียงตัวเดียว และจะต้องทำการรัน Kafka Connect หลายๆ เซสชั่นเพื่อรองรับการทำงานแบบ parallel
• connect-mirror : ใช้สำหรับทำ Mirror ข้อมูลจาก Kafka Cluster หรือ Topic ไปยัง Kafka Cluster หรือ Topic อื่นๆ โดย Mirror คือการทำซ้ำ หรือ copy ข้อมูลจาก Kafka Cluster หรือ Topic ไปยังอีก Kafka Cluster หรือ Topic ที่ต่างกัน
• connect-distributed : เป็นโหมดการทำงานของ Kafka Connect ที่ใช้ worker หลายๆตัว โดยการทำงานของมันจะเป็นการสร้าง worker หลายๆตัว ทำงานแบบ distributed แต่ละตัวแบ่งงานกันทำการเชื่อมต่อและคอนฟิกกับ Kafka Broker เพื่อการนำเข้า/ส่งออกข้อมูล ทำให้สามารถ scale ระบบได้เพื่อรองรับการทำงานแบบ parallel หรือยืดหยุ่นกับการปรับปรุงประสิทธิภาพของระบบได้ง่ายขึ้น

ในการใช้งานครั้งนี้เราจะใช้ตัว connect-standalone ในการทดสอบการเชื่อมต่อไปยังต้นทาง mysql database(เก็บข้อมูลปกติ) และปลายทางเป็น mysql database(สำหรับเก็บข้อมูลจาก topic จาก kafka) โดยเราจะfocus ไปที่การตั้งค่าในไฟล์ config ที่ใช้สำหรับเชื่อมต่อโดนเริ่มจากการ config สำหรับ mysql(เก็บข้อมูลปกติ)

Step 1. ติดตั้ง Debezium

โดยเราจะทำการ download file debezium มาที่เครื่องโดยใช้คำสั่ง

wget https://repo1.maven.org/maven2/io/debezium/debezium-connector-mysql/1.9.7.Final/debezium-connector-mysql-1.9.7.Final-plugin.tar.gz

tar -xzf debezium-connector-mysql-1.9.7.Final-plugin.tar.gz

ทำการย้าย folder ที่ได้มาด้วยคำสั่ง

mv debezium-connector-mysql /home/<ที่อยู่ directory ที่ติดตั้ง kafka>/kafka

Step 2. ติดตั้ง Confluent jdbc connector

ทำการ download https://www.confluent.io/hub/confluentinc/kafka-connect-jdbc ไว้ที่เครื่องและทำการแตกไฟล์ จากนั้นทำการย้ายไฟล์ไปไว้ใน directory เดียวกันกับ Debezium ด้วยคำสั่ง

mv confluentinc-kafka-connect-jdbc-10.6.3 /home/<ที่อยู่ directory ที่ติดตั้ง kafka>/kafka

Step 3. ติดตั้ง connector ของ mysql

ทำการ download ไฟล์จาก https://dev.mysql.com/downloads/connector/j/ มาไว้ที่เครื่องจากนั้นจะทำการย้ายไฟล์ .jar โดยใช้คำสั่ง

mv mysql-connector-j_8.0.32-1ubuntu22.04_all.deb /home/<ที่อยู่ directory ที่ติดตั้ง kafka>/kafka

Step 4. ตั้งค่า Connector สำหรับเชื่อมต่อ mysql(เก็บข้อมูลปกติ)

เข้าไปที่ directory /config และทำการสร้างไฟล์ properties โดยใช้คำสั่ง

nano mysql-connector.properties

โดยภายในไฟล์ mysql-connector.properties จะประกอบไปด้วย

name=sql-connect
connector.class=io.debezium.connector.mysql.MySqlConnector
tasks.max=1
database.hostname=<ip เครื่อง mysql ที่เราต้องการ capture change>
database.port=3306
database.user=<username ของเครื่อง mysql>
database.password=<password ของเครื่อง mysql>

database.server.id=<ID ของ MySQL server สามารถกำหนดเองได้>
database.server.name=<ชื่อของ MySQL server สามารถกำหนดเองได้>

database.history.kafka.bootstrap.servers=<ip เครื่อง kafka ทำการบันทึกการกระทำต่างๆที่เกิดขึ้นใน database>

database.history.kafka.topic=<ชื่อ topic ของ Kafka ที่จะใช้เพื่อเก็บข้อมูลจาก database>
database.whitelist=<ชื่อของ database ที่ต้องการให้ Connector เชื่อมต่อ *สามารถระบุได้มากกว่า 1 schema>
database.include.list=<ชื่อของ database ที่ต้องการให้ Connector เชื่อมต่อ *สามารถระบุได้มากกว่า 1 schema>

transforms=route
transforms.route.type=org.apache.kafka.connect.transforms.RegexRouter
transforms.route.regex=([^.]+)\\.([^.]+)\\.([^.]+)
transforms.route.replacement=$3

ransforms = unwrap,dropTopicPrefix
transforms.unwrap.type = io.debezium.transforms.ExtractNewRecordState
transforms.unwrap.drop.tombstones = false
delete.handling‹.mode = none
transforms.dropTopicPrefix.type = org.apache.kafka.connect.transforms.RegexRouter
transforms.dropTopicPrefix.regex = <server name>.<table name>.(.*)
transforms.dropTopicPrefix.replacement = $1
include.schema.changes : true

Step 5. Connector สำหรับเชื่อมต่อ mysql(ที่ใช้เก็บข้อมูลจาก kafka)

nano jdbc-connector.properties

โดยภายในไฟล์ jdbc-connector.properties จะประกอบไปด้วย

name=jdbc-connector
connector.class=io.confluent.connect.jdbc.JdbcSinkConnector
tasks.max=1
topics=<ชื่อ topic>
connection.url=jdbc:mysql://<ip เครื่อง mysql พร้อม port>/<ชื่อ schema ที่จะใช้เก็บข้อมูลจาก kafka>
connection.user=<username ของเครื่อง mysql>
connection.password=<password ของเครื่อง mysql>
transforms=unwrap
transforms.unwrap.type=io.debezium.transforms.ExtractNewRecordState
transforms.unwrap.drop.tombstones=false
auto.create=true
delete.enabled=true
insert.mode=upsert
pk.mode=record_key
pk.fields=<กำหนด fields ที่จะใช้เป็น primary key>

สำหรับ pk.mode เป็นการกำหนดวิธีการจัดการกับ Primary Key ของตารางในฐานข้อมูล โดยมีค่าที่สามารถกำหนดได้ดังนี้
none - ไม่ใช้ Primary Key ในการจัดการข้อมูล
record_key - ใช้ค่า key ของ Kafka record เป็น Primary Key ในการจัดการข้อมูล
record_value - ใช้ค่า value ของ Kafka record เป็น Primary Key ในการจัดการข้อมูล
record_headers - ใช้ค่า headers ของ Kafka record เป็น Primary Key ในการจัดการข้อมูล
custom - ใช้ฟิลด์ที่กำหนดเองเป็น Primary Key ในการจัดการข้อมูล
โดยค่า default คือ none

ส่วนค่า insert.mode เป็นการกำหนดวิธีการจัดการข้อมูลที่ถูกส่งมาจาก Kafka topic ไปยังฐานข้อมูล โดยมีค่าที่สามารถกำหนดได้ดังนี้
insert - จะทำการ insert ข้อมูลใหม่เมื่อไม่พบ Primary Key หรือเมื่อพบ Primary Key แต่ไม่มีข้อมูลในตาราง
upsert - จะทำการ update ข้อมูลเมื่อพบ Primary Key และมีข้อมูลอยู่ในตาราง หรือทำการ insert ข้อมูลใหม่เมื่อไม่พบ Primary Key หรือไม่มีข้อมูลในตาราง
update - จะทำการ update ข้อมูลเมื่อพบ Primary Key และมีข้อมูลอยู่ในตาราง แต่ไม่ทำการ insert ข้อมูลใหม่เมื่อไม่พบ Primary Key หรือไม่มีข้อมูลในตาราง
delete - จะทำการ delete ข้อมูลเมื่อพบ Primary Key
โดยค่า default คือ insert หากไม่ระบุ insert.mode ในการกำหนดค่าใหม่.

Step 6. ตั้งค่า plugin

ทำการแก้ไขไฟล์ connect-standalone.properties ด้วยคำสั่ง

nano connect-standalone.properties

ทำการเพิ่ม path directory ของ plugin ในบรรทัด

plugin.path=/home/kafka/Downloads , /usr/share, /usr/bin/java , /home/<kafka_install>/kafka_2.13-3.4.0

Step 7. Start connector

ให้เราเข้าไปที่ path /bin และใช้คำสั่งต่อไปนี้เพื่อเริ่มการทำงานของ connector

./connect-standalone.sh ../config/connect-standalone.properties ../config/mysql-connector.properties ../config/jdbc-connector.properties

ตรวจสอบ console ว่าพบ error หรือไม่หากไม่พบแปลว่าตอนนี้เราสามารถเชื่อมต่อเข้ากับ mysql ทั้งสองฝั่งได้เรียบร้อยแล้ว

*กรณีที่เราต้องการต่อ jdbc เข้ากับตัว PostgreSQL สามารถใช้ค่า config คล้ายๆกับตอนเชื่อต่อไปยัง mysql ได้เลย

name=<ชื่อ conector>
connector.class=io.confluent.connect.jdbc.JdbcSinkConnector
tasks.max=1
topics=<ชื่อ topic>
connection.url=jdbc:postgresql://<ip เครื่อง postgre>/<database name>
connection.user=<username>
connection.password=<password>
transforms=unwrap
transforms.unwrap.type=io.debezium.transforms.ExtractNewRecordState
transforms.unwrap.drop.tombstones=false
auto.create=true
delete.enabled=true
insert.mode=upsert
pk.mode=record_key
pk.fields=<กำหนด primary key>

• เครื่อง Ubuntu (ณ ตอนที่เขียนใช้ version 22.04)
• ในการลงจะเป็นการติดตั้งผ่านตัว terminal ภายในเครื่อง แนะนำให้ใช้งานด้วย root user

Step 1. Setup machine

ทำการเช็คความพร้อมของตัวเครื่องที่เราจะลง kafka เริ่มจากการตรวจสอบ version ของ java ในตัวเครื่องว่าได้มีการลง java ไปแล้วหรือยังด้วยคำสั่ง

java -version

จากนั้นทำการ update ตัว apt-get ด้วยคำสั่ง

sudo apt-get update

หลังจากที่ได้ทำการ update เรียบร้อยแล้วและภายในเครื่องยังไม่ได้มีการติดตั้ง java ไว้เราจะใช้คำสั่ง

sudo apt install default-jdk

เพื่อทำการติดตั้ง java ลงไปที่เครื่อง ubuntu ของเรา

Step 2. Download & Setup

ทำการ download kafka จาก https://kafka.apache.org/downloads และทำการเลือก download ไฟล์นามสกุล kafka_2.12–3.x.0.tgz (สามารถใช้ web browser ในเครื่อง download ได้เลย)

โดยหลังจาก download เสร็จให้เราเข้าไปที่ Dowload directory ที่เก็บไฟล์ไว้ผ่านทาง terminal และทำการแตกไฟล์โดยใช้คำสั่ง

tar xzf kafka_2.12-3.4.0.tgz

จากนั้นทำการย้าย folder ของ kafka ไปยัง directory อื่นๆด้วยคำสั่ง

mv kafka_2.12-3.4.0 /home/<ตำแหน่ง directory ที่เราต้องการนำ folder ไปวาง>

Step 3. Config

ให้เราเข้าไปยัง folder ของ kafka ที่เราได้ย้ายไปและเข้าไปใน directory config

cd <directory ที่เราย้ายไฟล์มา>/kafka_2.12-3.4.0/config

จากนั้นเราจะทำการตั้งค่าตัว server.properties โดยใช้คำสั่ง

nano server.properties

ทำการลบ comment บรรทัดที่มีค่า “listeners=PLAINTEXT://localhost:9092” และทำการเพิ่ม “advertised.host.name=localhost” เข้าไปอีก 1 บรรทัด แล้วกด ctrl+x เพื่อออกจากหน้าจอแก้ไข

Step 4. Start

หลังจากที่ทำการ config เรียบร้อยแล้วให้เราย้ายไปที่ directory /bin

cd <kafka location>/kafka_2.12-3.4.0/bin

และจะเริ่มทำการ start ตัว zookeeper ด้วยคำสั่ง

.zookeeper-server-start.sh config/zookeeper.properties

หลังจาก start zookeeper เรียบร้อยไม่มี error เราทำการ start ตัว kafka โดยจะทำการเปิด terminal เพิ่มอีก 1 หน้าต่างแล้วทำการเข้ามายัง directory /bin และใช้คำสั่ง start kafka

.kafka-server-start.sh config/server.properties

ทำการ check start log ว่ามี error อะไรหรือไม่

Step 5. Test

หลังจากที่เรา start kafka ได้สำเร็จแล้วจะทำการทดลองใช้งานกันดูโดยเปิด terminal หน้าต่างใหม่ขึ้นมา แล้วทำการเข้าไปยัง directory /bin

คำสั่งสำหรับ check topic ในเครื่อง

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

คำสั่งสำหรับสร้าง topic

.kafka-topics.sh --create --topic <ชื่อ topic> --bootstrap-server localhost:9092

คำสั่งสำหรับส่ง message เข้าไปยัง topic ผ่านตัว kafka-console

./kafka-console-producer.sh --topic <ชื่อ topic> --bootstrap-server localhost:9092

คำสั่งสำหรับแสดง message ที่อยู่ใน topic

./kafka-console-consumer.sh --topic <ชื่อ topic> --bootstrap-server localhost:9092 --from-beginning