Migrasi data - PolarDB - Alibaba Cloud Documentation Center

Panduan ini menyediakan langkah-langkah terperinci dan praktik terbaik untuk migrasi data dari Amazon DynamoDB ke PolarDB for PostgreSQL. PolarDB menawarkan rangkaian alat migrasi khusus yang menggabungkan sinkronisasi penuh dan inkremental untuk membantu Anda mencapai migrasi data yang lancar dengan downtime minimal.

Ikhtisar migrasi

Migrasi terdiri dari lima tahap, yang diatur oleh tiga alat inti: nimo-shake (untuk sinkronisasi data, termasuk sinkronisasi penuh dan inkremental), nimo-full-check (untuk pemeriksaan konsistensi data), dan PolarDBBackSync (untuk sinkronisasi balik).

数据迁移指南-流程图

Sinkronisasi penuh
- Alat: nimo-shake
- Proses: Alat ini secara otomatis membuat skema tabel yang sesuai di kluster PolarDB target. Kemudian, alat ini secara konkuren membaca semua data dari database sumber menggunakan operasi Scan dan menulisnya ke kluster target dalam batch menggunakan BatchWriteItem.
Sinkronisasi inkremental
- Alat: nimo-shake
- Proses: Setelah sinkronisasi penuh selesai, alat ini menggunakan AWS DynamoDB Streams untuk menangkap perubahan data (penyisipan, penghapusan, dan modifikasi) dari sumber sejak migrasi dimulai. Alat ini kemudian menyinkronkan perubahan tersebut ke kluster target untuk memastikan konsistensi data akhir. Proses ini mendukung transfer yang dapat dilanjutkan kembali.
Pemeriksaan konsistensi
- Alat: nimo-full-check
- Proses: Anda dapat menjalankan alat ini kapan saja selama atau setelah sinkronisasi data. Alat ini secara konkuren membaca data dari sumber dan target, membandingkannya berdasarkan primary key, dan menghasilkan laporan perbedaan terperinci untuk memverifikasi integritas data.
(Opsional) Sinkronisasi balik
- Alat: PolarDBBackSync.jar (berbasis Realtime Compute for Apache Flink)
- Proses: Setelah memverifikasi konsistensi data, Anda dapat membuat tautan sinkronisasi balik dari PolarDB for PostgreSQL ke DynamoDB sumber sebagai perlindungan untuk rollback bisnis. Alat ini menggunakan Flink untuk menangkap perubahan data real-time dari PolarDB dan memanggil operasi API DynamoDB PutItem atau DeleteItem untuk menyinkronkan pembaruan tersebut.
Cutover bisnis
- Proses: Ketika latensi data inkremental minimal dan pemeriksaan konsistensi tidak menunjukkan perbedaan, Anda dapat menghentikan sementara penulisan bisnis. Setelah semua data tersinkronisasi, alihkan koneksi aplikasi Anda ke kluster PolarDB untuk menyelesaikan migrasi.

Catatan

Dampak performa: Migrasi data, terutama selama sinkronisasi penuh, memberikan beban baca dan tulis yang tinggi pada database sumber dan target. Kami merekomendasikan melakukan migrasi selama jam sepi dan mengevaluasi kapasitas database Anda terlebih dahulu.
Konfigurasi keamanan: Sebelum cutover bisnis, batasi izin tulis pada kluster PolarDB target. Berikan akses tulis hanya kepada akun sinkronisasi data untuk mencegah korupsi data yang tidak disengaja.

Prasyarat

Sebelum memulai migrasi, pastikan Anda telah menyelesaikan persiapan berikut:

Dapatkan toolkit:
1. Toolkit migrasi: NimoShake_v20260616.zip. Paket ini berisi tiga alat: nimo-shake, nimo-full-check, dan nimo-repair.
2. (Opsional) Toolkit migrasi balik: PolarDBBackSync.jar.
Kluster PolarDB:
1. Untuk kluster yang sudah ada atau baru, aktifkan kompatibilitas DynamoDB, dapatkan titik akhir DynamoDB, dan buat akun DynamoDB khusus untuk mendapatkan kredensial akses guna akses API.
2. (Opsional) Konfigurasi parameter: Jika Anda perlu mengonfigurasi sinkronisasi balik, Anda harus mengubah parameter wal_level kluster PolarDB menjadi logical. Mengubah parameter ini memerlukan restart kluster, sehingga kami merekomendasikan menyelesaikan langkah ini sebelum memulai migrasi.
AWS DynamoDB:
1. Dapatkan kredensial akses (ID AccessKey dan Secret Access Key) untuk AWS DynamoDB.
Lingkungan runtime: Siapkan instance ECS atau server lain yang memiliki konektivitas ke kluster PolarDB dan AWS DynamoDB untuk menjalankan toolkit migrasi.

Prosedur

Langkah 1: Konfigurasikan dan mulai sinkronisasi data

Ekstrak NimoShake_v20260616.zip, masuk ke direktori NimoShake, dan edit file konfigurasi bersama conf/nimo.conf. Alat nimo-shake, nimo-full-check, dan nimo-repair berbagi file ini, dan masing-masing alat hanya membaca parameter yang diperlukannya. Berikut adalah parameter konfigurasi inti yang diperlukan untuk memulai sinkronisasi:

Parameter	Deskripsi	Nilai contoh
`sync_mode`	Mode sinkronisasi: `all` menunjukkan sinkronisasi penuh dan inkremental, sedangkan `full` hanya sinkronisasi penuh.	`all`
`source.access_key_id`	ID AccessKey dari AWS DynamoDB sumber.	`AKIAIOSFODNN7...`
`source.secret_access_key`	Secret access key dari AWS DynamoDB sumber.	`wJalrXUtnFEMI...`
`source.region`	Wilayah tempat AWS DynamoDB sumber berada.	`cn-north-1`
`target.endpoint_url`	Titik akhir akses DynamoDB dari kluster PolarDB target (termasuk nomor port).	`http://pe-xxx.rwlb.rds...`
`target.access_key_id`	ID AccessKey dari akun DynamoDB untuk kluster PolarDB target.	`your-polardb-access-key`
`target.secret_access_key`	Secret access key dari akun DynamoDB untuk kluster PolarDB target.	`your-polardb-secret-key`
`filter.collection.white`	Daftar putih untuk pemfilteran tabel. Pisahkan beberapa nama tabel dengan titik koma (`;`). Parameter ini tidak dapat digunakan bersama parameter daftar hitam.	`c1;c2`
`filter.collection.black`	Daftar hitam untuk pemfilteran tabel. Pisahkan beberapa nama tabel dengan titik koma (`;`). Parameter ini tidak dapat digunakan bersama parameter daftar putih.	`c1;c2`

Untuk menggunakan mode pemeriksaan berbasis snapshot S3 yang direkomendasikan (lihat Langkah 2), Anda juga harus mengonfigurasi parameter koneksi PostgreSQL native berikut:

Parameter	Deskripsi	Nilai contoh
`s3.export_state_file`	Jalur ke file status ekspor. File ini ditulis oleh `nimo-shake` dan dibaca oleh `nimo-full-check`.	`../nimo-shake-s3-exports.json`
`s3.export_bucket`	Nama bucket S3 target untuk ekspor data DynamoDB.	`my-export-bucket`
`s3.export_prefix`	Awalan jalur S3 yang digunakan untuk mengatur file yang diekspor.	`exports/my-project/`
`target.pg.endpoint_url`	Alamat port native PolarDB untuk PostgreSQL (format `host:port`).	`pc-xxx.pg.polardb.rds...:5432`
`target.pg.user`	Username untuk PolarDB untuk PostgreSQL (juga digunakan sebagai nama skema). Catatan Akun DynamoDB pada dasarnya adalah akun database standar. Secret access key-nya (SK) dihasilkan dengan mengenkripsi password yang ditentukan pengguna menggunakan algoritma yang kompatibel. Oleh karena itu, `target.pg.user` dan `target.access_key_id` merujuk pada akun yang sama. `target.pg.password` adalah password yang ditentukan pengguna, dan SK diturunkan dari password ini.	`your-username`
`target.pg.password`	Password untuk PolarDB untuk PostgreSQL.	`your-password`

(Opsional) Mode tulis langsung PG untuk sinkronisasi penuh

Secara default, sinkronisasi penuh menggunakan mode tulis langsung PostgreSQL native untuk menulis data langsung ke PolarDB menggunakan protokol PostgreSQL standar, melewati lapisan API kompatibilitas DynamoDB demi performa yang lebih tinggi. Untuk kembali ke lapisan API DynamoDB (BatchWriteItem, yang menulis maksimal 25 item sekaligus), Anda dapat mengubah parameter ini secara manual menjadi dynamodb. Anda dapat mengonfigurasi parameter berikut di bagian [SHAKE] file conf/nimo.conf:

Parameter	Deskripsi	Default
`full.write_protocol`	Protokol tulis penuh. `postgresql` (default) adalah mode berkinerja-tinggi untuk tulis langsung ke PostgreSQL. `dynamodb` adalah lapisan API DynamoDB (BatchWriteItem, maksimal 25 item per batch).	`postgresql`
`full.document.write.batch`	Jumlah dokumen yang ditulis dalam setiap batch. Saat Anda menggunakan tulis langsung ke PG, ukuran batch tidak dibatasi hingga 25 dokumen. Anda dapat meningkatkan nilai ini untuk lebih meningkatkan throughput tulis. Jika Anda kembali ke protokol `dynamodb`, ubah nilai ini kembali menjadi 25.	`1000`
`full.document.concurrency`	Jumlah thread tulis konkuren per tabel. Dalam mode tulis langsung PG, alat secara otomatis mengatur ukuran kolam koneksi berdasarkan nilai ini untuk memastikan korespondensi satu-ke-satu antara thread dan koneksi.	`4`

Catatan

Mode tulis langsung PG juga mengharuskan Anda mengonfigurasi tiga parameter target.pg.*. Kumpulan konfigurasi ini dibagi dengan mode pemeriksaan s3/incr, sehingga Anda tidak perlu memasukkannya lagi.
Mode ini hanya memengaruhi jalur tulis selama fase sinkronisasi penuh. Sinkronisasi inkremental selalu menulis data melalui lapisan API yang kompatibel dengan DynamoDB.

Jalankan file biner untuk sistem operasi Anda untuk memulai sinkronisasi.
Catatan
Kami merekomendasikan menjalankan tugas dengan nohup di latar belakang untuk mencegah tugas sinkronisasi terganggu jika sesi terminal terputus.
```
# Jalankan di direktori NimoShake
nohup ./bin/nimo-shake.linux.amd64 -conf=./conf/nimo.conf > /dev/null 2>&1 &
```
Program pertama kali melakukan sinkronisasi penuh. Setelah sinkronisasi penuh selesai, program secara otomatis beralih ke fase sinkronisasi inkremental dan terus berjalan.

Langkah 2: Periksa konsistensi data

nimo-full-check dan nimo-shake berbagi file conf/nimo.conf yang sama, sehingga Anda tidak perlu mengganti direktori atau mengedit file konfigurasi tambahan. Pilih mode pemeriksaan menggunakan parameter mode:

scan: Melakukan pemindaian real-time untuk membandingkan data antara sumber dan target. Mode ini paling mudah dikonfigurasi dan tidak memerlukan infrastruktur tambahan, tetapi terus-menerus mengonsumsi kapasitas baca DynamoDB selama pemeriksaan.
s3 (Direkomendasikan): Saat sinkronisasi penuh selesai, nimo-shake secara otomatis mengekspor data sumber ke S3 dan membuat snapshot sesaat. nimo-full-check menggunakan snapshot ini sebagai garis dasar untuk pemeriksaan, sehingga menghindari pemindaian ulang DynamoDB sumber. Ini memerlukan parameter terkait S3 dari Langkah 1.
incr (Gunakan bersama mode s3): Selama fase sinkronisasi inkremental, nimo-shake menulis primary key dari setiap catatan yang berubah ke tabel pemeriksaan (ct_{user}_{table}) di PolarDB. nimo-full-check membaca tabel pemeriksaan ini dan hanya memeriksa catatan yang berubah.

Perbandingan dua metode pemeriksaan

	Scan Mode	Mode S3 / incr
Downtime yang diperlukan	Lebih lama. Penulisan bisnis harus dihentikan selama pemeriksaan untuk mencegah data sumber berubah selama pemindaian.	Lebih singkat. Pemeriksaan berdasarkan snapshot. Setelah menghentikan penulisan, Anda hanya perlu memastikan bahwa data inkremental telah tersinkronisasi sebelum cutover.
Dampak pada sumber	Terus-menerus mengonsumsi kapasitas baca DynamoDB.	Pemeriksaan penuh berdasarkan snapshot S3 dan tidak mengonsumsi kapasitas DynamoDB. Pemeriksaan inkremental (mode incr) hanya mengkueri catatan yang berubah, sehingga mengonsumsi kapasitas lebih sedikit.
Kompleksitas konfigurasi	Rendah.	Lebih tinggi. Memerlukan konfigurasi tambahan untuk S3 dan koneksi PolarDB untuk PostgreSQL.

Mulai tugas pemeriksaan.
```
nohup ./bin/nimo-full-check.linux.amd64 -conf=./conf/nimo.conf > /dev/null 2>&1 &
```
Alat pemeriksaan menyimpan log terperinci dan laporan perbedaan data di direktori default logs/ dan nimo-full-check-diff/, masing-masing.
(Opsional) Perbaiki data yang tidak konsisten:
1. Saat data di DynamoDB sumber tidak konsisten dengan data di kluster PolarDB for PostgreSQL target, tabel yang tidak konsisten tercantum di direktori diff yang ditentukan. Contoh:
  Catatan
  Direktori diff default adalah nimo-full-check-diff. Anda dapat mengonfigurasinya di conf/nimo.conf menggunakan parameter diff_output_file.
```
nimo-full-check-diff/
  └── testtable-0
  └── testtable-1
```
2. Jalankan tugas perbaikan:
```
nohup ./bin/nimo-repair.linux.amd64 -conf=./conf/nimo.conf > /dev/null 2>&1 &
```

Langkah 3: (Opsional) Konfigurasikan sinkronisasi balik

Sebelum cutover bisnis, Anda dapat mengonfigurasi tautan sinkronisasi balik dari PolarDB for PostgreSQL ke DynamoDB sumber. Tautan ini diaktifkan selama pergantian resmi ke PolarDB untuk mengaktifkan aliran data balik dan menyediakan perlindungan untuk rollback bisnis. Prosedur konfigurasinya sebagai berikut:

Persiapan lingkungan

Pastikan parameter PolarDB wal_level diatur ke logical.
Buat akun istimewa dan berikan izin:
1. (Opsional) Jika Anda belum membuat akun istimewa, buka Konsol PolarDB, lalu di halaman kluster Settings and Management > Accounts, buat akun istimewa.

Buat slot replikasi logis dan berikan izin: Hubungkan ke database polardb_internal_dynamodb menggunakan akun istimewa. Jalankan perintah SQL berikut untuk membuat slot replikasi logis dan memberikan izin replikasi kepada DynamoDB Account. Anda kemudian dapat melihat status slot replikasi logis yang dibuat.

-- Buat slot replikasi logis. Nama 'flink_slot' harus sesuai dengan nama slot yang ditentukan dalam konfigurasi Flink berikutnya.
SELECT * FROM pg_create_logical_replication_slot('flink_slot', 'pgoutput');

-- Berikan izin REPLICATION kepada akun pengguna DynamoDB khusus yang Anda buat sebelumnya.
-- Ganti <your_dynamodb_user> dengan nama akun pengguna DynamoDB khusus Anda.
ALTER ROLE <your_dynamodb_user> REPLICATION;

-- Periksa status slot replikasi. Pada tahap ini, status aktif 'flink_slot' seharusnya f (false).
SELECT * FROM pg_replication_slots;

Aktifkan dan konfigurasikan Flink:
1. Aktifkan Realtime Compute for Apache Flink dan buat ruang kerja Flink.
  Penting
  Ruang kerja Flink harus berada di VPC yang sama dengan kluster PolarDB.
2. Konfigurasikan akses jaringan publik untuk ruang kerja Flink Anda agar dapat terhubung ke AWS DynamoDB.
Konfigurasikan daftar putih alamat IP kluster PolarDB:
1. Di Konsol Flink, klik tombol View Details untuk ruang kerja, lalu di halaman Workspace Details, dapatkan informasi CIDR Block-nya.
2. Buka Konsol PolarDB. Pada halaman kluster, pilih Settings and Management > Cluster Whitelists, lalu klik Add Whitelist dan tambahkan blok CIDR Flink.
Verifikasi konektivitas antara kluster PolarDB dan ruang kerja Flink:
1. Di Konsol Flink, buka ruang kerja dan klik ikon Network Probe di pojok kanan atas.
2. Masukkan titik akhir pribadi dan port dari node primer kluster PolarDB, lalu klik Probe.
3. Pesan pop-up yang bertuliskan Network Probe Connection Successful menunjukkan bahwa daftar putih alamat IP kluster telah dikonfigurasi dengan benar.

Terapkan pekerjaan Flink

Unduh alat sinkronisasi balik: PolarDBBackSync.jar.

Buat file konfigurasi bernama application.yaml dengan konten berikut:

snapshot:
  mode: never

source:
    # Titik akhir pribadi dari node primer PolarDB
    hostname: pc-xxx.pg.polardb.rds.aliyuncs.com
    # Port pribadi dari node primer PolarDB
    port: 5432
    # Nama slot replikasi logis yang dibuat sebelumnya
    slotName: flink_slot

target:
  # Wilayah AWS DynamoDB target
  region: cn-north-1

# (Opsional) Konfigurasi pemfilteran tabel. Anda dapat mendeklarasikan whiteTableSet atau blackTableSet.
filter:
  # whiteTableSet: tabel yang akan disinkronisasi balik
  # blackTableSet: tabel yang tidak akan disinkronisasi balik
  whiteTableSet:
  blackTableSet:

# Interval checkpoint untuk pekerjaan Flink, dalam milidetik
checkpoint:
  interval: 3000

Unggah file: Di Konsol Flink, buka ruang kerja target. Di halaman File Management, unggah PolarDBBackSync.jar dan application.yaml.

Simpan kredensial secara aman: Untuk menghindari eksposur kunci dalam teks biasa, kami merekomendasikan menggunakan fitur Variable Management Flink untuk menyimpan informasi sensitif. Di halaman Variable Management, tambahkan empat variabel berikut:

Parameter	Nilai
`polardbusername`	Akun DynamoDB untuk PolarDB.
`polardbpassword`	Password asli dari akun DynamoDB untuk PolarDB. Ini bukan secret access key akun DynamoDB.
`dynamodbak`	ID AccessKey AWS DynamoDB.
`dynamodbsk`	Secret access key AWS DynamoDB.
`configfilename`	(Opsional) Nama file dependensi tambahan. Nilai default: `application.yaml`.

Terapkan dan mulai pekerjaan:

Buka halaman Job O&M, lalu pilih Deploy Job > JAR Job.

Masukkan parameter utama berikut dan konfigurasikan parameter lain berdasarkan lingkungan bisnis Anda. Lalu, klik Deploy.

Parameter	Referensi
Deployment Mode	Atur ke Stream Mode.
Deployment name	Masukkan nama untuk penerapan pekerjaan, misalnya PolarDBBackSync.
Engine Version	Atur ke vvr-11.3-jdk11-flink-1.20.
JAR URI	Pilih `PolarDBBackSync.jar` yang diunggah.
Entry point class	Atur ke `org.example.PolarDBCdcJob`.
Entry point main arguments	Atur ke: `--polardbusername ${secret_values.polardbusername}` `--polardbpassword ${secret_values.polardbpassword}` `--dynamodbak ${secret_values.dynamodbak}` `--dynamodbsk ${secret_values.dynamodbsk}` (Opsional) `--configfilename ${secret_values.configfilename}`
Additional dependency files	Pilih `application.yaml` yang diunggah.

Penting

Jika password atau nilai parameter lain berisi karakter khusus, pekerjaan Flink mungkin gagal mengurai parameter. Untuk mencegah masalah ini, setelah pekerjaan dibuat, buka Deployment Details > Run Parameter Configuration, klik Edit, dan tambahkan konfigurasi berikut di Other Configurations: env.java.opts: -Dconfig.disable-inline-comment=true.

Setelah penerapan berhasil, klik Start > Stateless Start.

Validasi dan pembersihan

Validasi: Setelah pekerjaan dimulai, gunakan akun istimewa untuk terhubung ke database polardb_internal_dynamodb kluster PolarDB dan jalankan SELECT * FROM pg_replication_slots;. Jika nilai bidang active untuk slot replikasi flink_slot berubah menjadi t (true), pekerjaan Flink telah terhubung. Anda kemudian dapat mulai mengarahkan trafik bisnis ke kluster PolarDB.
Pembersihan: Saat sinkronisasi balik tidak lagi diperlukan, Anda dapat melakukan langkah-langkah berikut untuk melepaskan sumber daya terkait dan mengurangi biaya.
- Realtime Compute for Apache Flink:
  - Hentikan pekerjaan: Buka Konsol Flink. Di ruang kerja target, buka halaman Job O&M, temukan pekerjaan target, lalu klik Stop.
  - Lepaskan instans: Kembali ke Konsol Flink, temukan ruang kerja target, lalu klik Release Resources.
- Kluster PolarDB: Gunakan akun istimewa untuk terhubung ke database polardb_internal_dynamodb dan jalankan perintah berikut untuk menghapus slot replikasi logis.
```
SELECT pg_drop_replication_slot('flink_slot');
```

Langkah 4: Lakukan cutover bisnis

Setelah latensi sinkronisasi inkremental rendah dan pemeriksaan konsistensi data berhasil, Anda dapat merencanakan cutover bisnis. Saat siap, ikuti langkah-langkah berikut:

Pemeriksaan akhir: Sebelum jendela downtime yang direncanakan, jalankan alat pemeriksaan konsistensi berulang kali hingga Anda memastikan bahwa latensi sinkronisasi inkremental minimal dan jumlah ketidaksesuaian data nol atau dalam kisaran yang dapat diterima.
Hentikan penulisan sumber: Pada awal jendela downtime, hentikan semua aplikasi dari menulis ke AWS DynamoDB sumber.
Tunggu sinkronisasi selesai: Pantau log nimo-shake untuk memastikan tidak ada data inkremental baru yang perlu disinkronkan.
Pemeriksaan konsistensi terakhir: Jalankan alat nimo-full-check sekali lagi untuk memastikan data di sumber dan target sepenuhnya konsisten.
Hentikan alat sinkronisasi: Setelah memastikan data telah tersinkronisasi sepenuhnya, hentikan proses nimo-shake.
Alihkan koneksi aplikasi: Hentikan aplikasi bisnis dan ubah konfigurasi koneksi database mereka dari titik akhir AWS DynamoDB ke titik akhir akses DynamoDB PolarDB.
(Opsional) Mulai tugas sinkronisasi balik: Selama pergantian resmi bisnis ke PolarDB, mulai tugas sinkronisasi balik. Tugas ini menggunakan Realtime Compute for Apache Flink untuk menerapkan replikasi data balik, yang menyediakan perlindungan untuk rollback bisnis.
Mulai aplikasi bisnis: Mulai ulang aplikasi bisnis. Cutover selesai.
(Opsional) Hentikan tugas sinkronisasi balik: Setelah cutover selesai dan bisnis berjalan stabil selama periode tertentu, Anda dapat menghentikan tugas sinkronisasi balik (yang mencakup menghentikan pekerjaan Flink dan melepaskan sumber dayanya).

Lampiran: Simulasikan trafik real-time

Untuk mensimulasikan migrasi dunia nyata dengan penulisan data kontinu di lingkungan pengujian, gunakan kode contoh Go berikut. Kode ini secara berkala menulis dan memperbarui data di tabel DynamoDB sumber.

Catatan

Langkah ini hanya untuk pengujian dan validasi proses migrasi. Jangan lakukan langkah ini dalam migrasi produksi.

package main

import (
	"context"
	"fmt"
	"log"
	"math/rand"
	"time"

	"github.com/aws/aws-sdk-go-v2/aws"
	"github.com/aws/aws-sdk-go-v2/config"
	"github.com/aws/aws-sdk-go-v2/credentials"
	"github.com/aws/aws-sdk-go-v2/service/dynamodb"
	"github.com/aws/aws-sdk-go-v2/service/dynamodb/types"
)

// --- Konfigurasi untuk AWS DynamoDB sumber Anda ---
var (
    region    = "cn-north-1"       // Wilayah AWS DynamoDB
    accessKey = "your-aws-access-key" // Kunci akses AWS DynamoDB
    secretKey = "your-aws-secret-key" // Kunci rahasia AWS DynamoDB
)

// --- Fungsi pembantu untuk membuat klien DynamoDB ---
func createClient() (*dynamodb.Client, context.Context) {
    ctx := context.Background()
    sdkConfig, err := config.LoadDefaultConfig(ctx, config.WithRegion(region))
    if err != nil {
        log.Fatalf("Gagal memuat konfigurasi AWS: %v", err)
    }
    client := dynamodb.NewFromConfig(sdkConfig, func(o *dynamodb.Options) {
        o.Credentials = credentials.NewStaticCredentialsProvider(accessKey, secretKey, "")
    })
    return client, ctx
}

// --- Fungsi untuk membuat tabel dan mengisi data awal ---
func initializeData(client *dynamodb.Client, ctx context.Context) {
    tableName := "src1" // Nama tabel contoh

    // Buat tabel jika belum ada
    _, err := client.CreateTable(ctx, &dynamodb.CreateTableInput{
        TableName: &tableName,

        AttributeDefinitions: [ ]types.AttributeDefinition{

            {AttributeName: aws.String("pk"), AttributeType: types.ScalarAttributeTypeS},
        },

        KeySchema: [ ]types.KeySchemaElement{

            {AttributeName: aws.String("pk"), KeyType: types.KeyTypeHash},
        },
        ProvisionedThroughput: &types.ProvisionedThroughput{
            ReadCapacityUnits:  aws.Int64(100),
            WriteCapacityUnits: aws.Int64(100),
        },
    })
    if err != nil {
		// Abaikan jika tabel sudah ada, gagal untuk error lain
        if _, ok := err.(*types.ResourceInUseException); !ok {
			log.Fatalf("CreateTable gagal untuk %s: %v", tableName, err)
		}
    }

    fmt.Printf("Menunggu tabel '%s' menjadi aktif...\n", tableName)
    waiter := dynamodb.NewTableExistsWaiter(client)
    err = waiter.Wait(ctx, &dynamodb.DescribeTableInput{TableName: &tableName}, 5*time.Minute)
    if err != nil {
        log.Fatalf("Waiter gagal untuk tabel %s: %v", tableName, err)
    }

    // Masukkan 100 item contoh
    for i := 0; i < 100; i++ {
        pk := fmt.Sprintf("%s_user_%03d", tableName, i)
        item := map[string]types.AttributeValue{
            "pk":  &types.AttributeValueMemberS{Value: pk},
            "val": &types.AttributeValueMemberN{Value: fmt.Sprintf("%d", i)},
        }
        client.PutItem(ctx, &dynamodb.PutItemInput{TableName: &tableName, Item: item})
    }
    fmt.Printf("Memasukkan 100 item awal ke '%s'.\n", tableName)
}

// --- Fungsi untuk mensimulasikan trafik bisnis kontinu ---
func simulateTraffic(client *dynamodb.Client, ctx context.Context) {
    tableName := "src1"
    fmt.Println("Memulai pembaruan berkala untuk mensimulasikan trafik. Tekan Ctrl+C untuk berhenti.")
    i := 0
    for {
        pk := fmt.Sprintf("%s_user_%03d", tableName, i%100)
        newValue := fmt.Sprintf("%d", rand.Intn(1000))
        _, err := client.UpdateItem(ctx, &dynamodb.UpdateItemInput{
            TableName: &tableName,
            Key: map[string]types.AttributeValue{
                "pk": &types.AttributeValueMemberS{Value: pk},
            },
            ExpressionAttributeValues: map[string]types.AttributeValue{
                ":newval": &types.AttributeValueMemberN{Value: newValue},
            },
            UpdateExpression: aws.String("SET val = :newval"),
        })
        if err != nil {
            fmt.Printf("Error pembaruan: %v\n", err)
        } else {
            fmt.Printf("Memperbarui pk=%s dengan val baru=%s\n", pk, newValue)
        }
        i++
        time.Sleep(1 * time.Second) // Perbarui satu catatan per detik
    }
}

func main() {
    client, ctx := createClient()
    initializeData(client, ctx)
    simulateTraffic(client, ctx)
}