Luồng đồng bộ dữ liệu (Sync Data Flow)

Tổng quan

Dịch vụ GMAC MEO tự động thực hiện đồng bộ dữ liệu và scraping xếp hạng tìm kiếm hàng ngày theo lịch định kỳ. Dữ liệu master từ 3 hệ thống khách hàng (GCOR, GMAC/Mappy, PIPIT) được tập hợp vào scraper DB, sau đó thực hiện scraping song song bằng AWS Batch, và ghi kết quả ngược lại về từng hệ thống.

Các hệ thống liên quan

Hệ thống	system_id	DB Cluster	Mục đích
GMAC / Mappy	1	`gmac-db-production-cluster-1`	DB khách hàng chính
GCOR	2	`gcor-db-production-cluster`	DB khách hàng
PIPIT	3	`kingmeo-db-production-cluster`	DB khách hàng
Scraper	-	`scraper.cluster-*`	DB xử lý trung gian

Luồng tổng thể

Chi tiết từng bước

Step 1: sync_database

Mục	Giá trị
Tên Lambda	`sync_database`
Runtime	Python 3.13
Timeout	900 giây
Role	`Lambda_scraper`
VPC	Có (private subnet)
Thời gian chạy	~2 phút

Xử lý:

Kết nối đến 3 DB: GCOR, GMAC/Mappy, PIPIT
Tạo bảng backup cho dữ liệu hiện tại (backup_*)
Đồng bộ các bảng sau vào scraper DB:
- mappy_user_settings
- mappy_gbp_locations
- mappy_gbp_locations_store_codes
- mappy_integrated_gbp_locations
- mappy_keywords

Sau khi hoàn thành, phát custom event lên EventBridge:

json

{
  "detail-type": "SyncDatabaseCompleted",
  "source": "mappy.sync_db",
  "detail": {
    "status": "completed",
    "timestamp": "2026-04-07T17:02:13+09:00"
  }
}

Biến môi trường (NEXT_LAMBDA_FUNCTION_NAME): fetch_search_ranking — tên Lambda sẽ được gọi tiếp theo.

Step 2: fetch_search_ranking

Mục	Giá trị
Tên Lambda	`fetch_search_ranking`
Runtime	Python 3.13
Timeout	900 giây
Role	`TestCreateBatch-role-*`
Thời gian chạy	~47 giây

Xử lý:

Được trigger bởi EventBridge event SyncDatabaseCompleted
Submit 80 AWS Batch jobs:
- Batch 1~40: normal (tìm kiếm thuận) — mỗi batch 350 keyword, Offset 0~13,650
- Batch 41~80: reverse (tìm kiếm ngược) — mỗi batch 350 keyword, Offset 0~13,650
Tổng cộng ~28,000 keyword được xử lý song song

Step 3: AWS Batch (Fargate)

Mục	Giá trị
Job Definition	`scraper_def`
Job Queue	`scrap_queue`
Compute Environment	`scraper_env_v1` (Fargate)
Container Image	`881980194724.dkr.ecr.ap-northeast-1.amazonaws.com/scraper:latest`
vCPU	1.0
Memory	2048 MB
Số container song song	Tối đa 80
Thời gian chạy	~3-4 giờ

Xử lý:

Mỗi container dùng Playwright để thực hiện tìm kiếm Google và lấy dữ liệu xếp hạng
Kết quả được ghi vào scraper DB
Khi container hoàn thành, tăng counter total_batch_done

Step 4: reTry_Failed_Keywords

Mục	Giá trị
Tên Lambda	`reTry_Failed_Keywords`
Runtime	Python 3.14
Timeout	900 giây
Trigger	Được gọi mỗi khi một Batch job hoàn thành
Thời gian chạy	~100ms (chưa đủ) / ~2 giây (khi đủ điều kiện)

Xử lý:

Được gọi bởi mỗi Batch container khi hoàn thành (callback)
Kiểm tra bảng mst_systems trong scraper DB: so sánh total_batch_done với total_batch (=80)
Nếu chưa đủ 80/80 → thoát ngay
Khi tất cả batch hoàn thành (80/80):
- Gọi lại fetch_search_ranking để retry các keyword thất bại

Step 5: syncResultFetchSearch

Mục	Giá trị
Tên Lambda	`syncResultFetchSearch`
Runtime	Python 3.14
Timeout	900 giây
Role	`Lambda_scraper`
Memory	5120 MB
Thời gian chạy	~46 giây

Xử lý:

Step 1: Đồng bộ kết quả tìm kiếm từ scraper DB về DB các hệ thống
- mappy_search_ranking_exports
- mappy_temp_search_rankings
Step 2: Đồng bộ dữ liệu process
- mappy_search_ranking_processes — cập nhật process_id
- Cập nhật process_id cho exports và temp_rankings liên quan
Step 2.5: Xử lý update logs
- mappy_search_ranking_update_logs — cập nhật record ngày hiện tại (status=1)
- Gán update_log_id cho các bảng liên quan
Hậu xử lý:
- Reset total_batch_done=0, retry_count=0 trong mst_systems
- Tăng run_times thêm 1

Timeline (ví dụ chạy hàng ngày)

08:00 JST  sync_database bắt đầu
08:02 JST  sync_database hoàn thành → Phát event SyncDatabaseCompleted
08:02 JST  fetch_search_ranking khởi động → Submit 80 Batch jobs
08:03 JST  AWS Batch bắt đầu chạy (80 container song song)
   ...     Mỗi batch done → gọi reTry_Failed_Keywords
12:00 JST  Tất cả batch hoàn thành (80/80)
12:00 JST  fetch_search_ranking (retry) khởi động
   ...     Retry Batch cho keyword thất bại
12:30 JST  syncResultFetchSearch chạy
12:31 JST  Đồng bộ kết quả hoàn thành → Reset counter

Luồng sau khi tích hợp Places API Batch

Tích hợp Places API batch vào luồng hiện tại. Places API batch chạy song song với scraping batch. Điều kiện để đồng bộ kết quả (syncResultFetchSearch) thay đổi: cả scraping (80/80) VÀ Places API đều phải hoàn thành.

Thay đổi bảng mst_systems

Cột	Kiểu	Mô tả
`total_batch`	int	Tổng số batch scraping (thay đổi tùy theo dữ liệu)
`total_batch_done`	int	Số batch scraping đã hoàn thành
`retry_count`	int	Số lần retry
`places_api_done`	int	(Mới) Cờ hoàn thành Places API batch (0=chưa, 1=xong)

Thay đổi so với luồng hiện tại

Places API batch chạy cùng luồng (sau khi fetch_search_ranking submit scraping batch)
Khi Places API batch hoàn thành → gọi Lambda event để update places_api_done = 1 trong mst_systems
reTry_Failed_Keywords thay đổi điều kiện kiểm tra:
- Hiện tại: chỉ check total_batch_done = total_batch
- Sau khi tích hợp: check total_batch_done = total_batch VÀ places_api_done = 1
Chỉ khi cả 2 điều kiện thỏa mãn → mới gọi syncResultFetchSearch để đồng bộ kết quả (bao gồm cả scraping + Places API) về 3 server

Sơ đồ luồng

Chi tiết xử lý

fetch_search_ranking: Submit cả scraping batch (80 jobs) và Places API batch
Scraping batch: 80 container Playwright chạy song song, mỗi container xong thì total_batch_done++
Places API batch (places_api_def): Lấy ranking qua Google Places API (Text Search)
- Container image: python310:latest
- Chỉ xử lý location có data_source=1
- Khi hoàn thành → gọi Lambda event update places_api_done = 1
reTry_Failed_Keywords (thay đổi logic):
- Được gọi mỗi khi một batch hoàn thành (callback)
- Check mst_systems: total_batch_done = total_batch VÀ places_api_done = 1
- Chưa đủ cả 2 → exit
- Đủ cả 2 → gọi syncResultFetchSearch
syncResultFetchSearch: Đồng bộ kết quả scraping + Places API từ scraper DB về 3 server (GCOR, GMAC, PIPIT)
Hậu xử lý: Reset total_batch_done=0, retry_count=0, places_api_done=0

Timeline (sau khi tích hợp)

08:00 JST  sync_database bắt đầu
08:02 JST  sync_database hoàn thành → SyncDatabaseCompleted
08:02 JST  fetch_search_ranking khởi động
           → Submit 80 Scraping Batch jobs
           → Submit Places API Batch job
08:03 JST  Scraping + Places API chạy song song
   ...     Mỗi batch done → gọi reTry_Failed_Keywords
           (check: total_batch_done=total_batch AND places_api_done=1)
11:00 JST  Places API Batch hoàn thành → Lambda update places_api_done=1
12:00 JST  Scraping Batch hoàn thành (80/80)
12:00 JST  Cả 2 điều kiện thỏa → syncResultFetchSearch chạy
12:01 JST  Đồng bộ kết quả (Scraping + Places API) → Reset counter

Danh sách Lambda

Tên Lambda	Vai trò	Timeout	Memory
`sync_database`	Đồng bộ master data	900s	128 MB
`fetch_search_ranking`	Submit Batch jobs	900s	128 MB
`reTry_Failed_Keywords`	Polling batch + retry	900s	128 MB
`syncResultFetchSearch`	Đồng bộ kết quả	900s	5120 MB
`checkBatchStatus`	Kiểm tra trạng thái batch	3s	128 MB
`cloudwatch-to-slack`	Gửi thông báo lỗi → Slack	-	-

Tài nguyên liên quan

ECR: 881980194724.dkr.ecr.ap-northeast-1.amazonaws.com/scraper:latest
Batch Queue: scrap_queue (Fargate)
Batch Compute Env: scraper_env_v1
CloudWatch Log Groups:
- /aws/lambda/sync_database
- /aws/lambda/fetch_search_ranking
- /aws/lambda/reTry_Failed_Keywords
- /aws/lambda/syncResultFetchSearch
- /aws/batch/job

Luồng đồng bộ dữ liệu (Sync Data Flow) ​

Tổng quan ​

Các hệ thống liên quan ​

Luồng tổng thể ​

Chi tiết từng bước ​

Step 1: sync_database ​

Step 2: fetch_search_ranking ​

Step 3: AWS Batch (Fargate) ​

Step 4: reTry_Failed_Keywords ​

Step 5: syncResultFetchSearch ​

Timeline (ví dụ chạy hàng ngày) ​

Luồng sau khi tích hợp Places API Batch ​

Thay đổi bảng mst_systems ​

Thay đổi so với luồng hiện tại ​

Sơ đồ luồng ​

Chi tiết xử lý ​

Timeline (sau khi tích hợp) ​

Danh sách Lambda ​

Tài nguyên liên quan ​

Luồng đồng bộ dữ liệu (Sync Data Flow)

Tổng quan

Các hệ thống liên quan

Luồng tổng thể

Chi tiết từng bước

Step 1: sync_database

Step 2: fetch_search_ranking

Step 3: AWS Batch (Fargate)

Step 4: reTry_Failed_Keywords

Step 5: syncResultFetchSearch

Timeline (ví dụ chạy hàng ngày)

Luồng sau khi tích hợp Places API Batch

Thay đổi bảng mst_systems

Thay đổi so với luồng hiện tại

Sơ đồ luồng

Chi tiết xử lý

Timeline (sau khi tích hợp)

Danh sách Lambda

Tài nguyên liên quan