سلام امیدوارم این کمکتون بکنه:
## راهحل: CDC + SCD Type 2
معماری پیشنهادی:
OLTP (با CDC فعال) → جداول Change → ETL Process → DW (با SCD Type 2)
### مراحل پیادهسازی:
## 1️⃣ خواندن تغییرات از CDC
-- مثال SQL Server
SELECT
__$operation, -- 1=Delete, 2=Insert, 3=Update-Before, 4=Update-After
__$start_lsn,
CustomerID,
CustomerName,
City
FROM cdc.dbo_Customer_CT
WHERE __$start_lsn > @last_processed_lsn
ORDER BY __$start_lsn
## 2️⃣ پیادهسازی در DW با SCD Type 2
-- جدول DW با SCD Type 2
CREATE TABLE DW.DimCustomer (
DW_CustomerKey INT IDENTITY(1,1) PRIMARY KEY,
CustomerID INT, -- Business Key
CustomerName NVARCHAR(100),
City NVARCHAR(50),
-- فیلدهای SCD
EffectiveDate DATETIME,
EndDate DATETIME,
IsCurrent BIT,
-- فیلدهای Audit
SourceOperation VARCHAR(10), -- INSERT, UPDATE, DELETE
LoadDate DATETIME
)
## 3️⃣ لاجیک ETL برای Update
-- هنگام دریافت UPDATE از CDC:
-- 1. رکورد قبلی رو ببند
UPDATE DW.DimCustomer
SET
EndDate = @ChangeDate,
IsCurrent = 0
WHERE
CustomerID = @CustomerID
AND IsCurrent = 1
-- 2. رکورد جدید وارد کن
INSERT INTO DW.DimCustomer (
CustomerID, CustomerName, City,
EffectiveDate, EndDate, IsCurrent,
SourceOperation, LoadDate
)
VALUES (
@CustomerID, @NewCustomerName, @NewCity,
@ChangeDate, '9999-12-31', 1,
'UPDATE', GETDATE()
)
## 4️⃣ پیادهسازی کامل با Python/SSIS
import pyodbc
def process_cdc_to_dw(last_lsn):
# 1. خواندن تغییرات از CDC
cdc_changes = read_cdc_changes(last_lsn)
for change in cdc_changes:
operation = change['__$operation']
if operation == 4: # Update-After
# بستن رکورد قبلی
close_current_record(change['CustomerID'])
# ایجاد رکورد جدید
insert_new_record(
customer_id=change['CustomerID'],
name=change['CustomerName'],
city=change['City'],
effective_date=change['ChangeDate'],
operation='UPDATE'
)
elif operation == 2: # Insert
insert_new_record(...)
elif operation == 1: # Delete
close_current_record(change['CustomerID'])
## 5️⃣ ابزارهای پیشنهادی:
| ابزار | مزیت |
|------|------|
| SSIS | Integration آسان با SQL Server CDC |
| Azure Data Factory | Cloud-native، Pipeline orchestration |
| Debezium | Open-source، پشتیبانی از چند DB |
| Stored Procedures | کنترل کامل، بدون ابزار اضافی |
## 6️⃣ نمونه Stored Procedure کامل:
`sql
CREATE PROCEDURE usp_ProcessCDC_To_DW
AS
BEGIN
DECLARE @LastLSN BINARY(10)
-- آخرین LSN پردازششده
SELECT @LastLSN = LastProcessedLSN FROM ETL.CDCControl
-- پردازش تغییرات
DECLARE @Changes TABLE (
Operation INT,
CustomerID INT,
CustomerName NVARCHAR(100),
City NVARCHAR(50),
ChangeLSN BINARY(10),
ChangeDate DATETIME
)
INSERT INTO @Changes
SELECT
__$operation,
CustomerID,
CustomerName,
City,
__$start_lsn,
sys.fn_cdc_map_lsn_to_time(__$start_lsn)
FROM cdc.dbo_Customer_CT
WHERE __$start_lsn > @LastLSN
-- اعمال تغییرات در DW
DECLARE @Op INT, @CustID INT, @Name NVARCHAR(100),
@City NVARCHAR(50), @ChangeDate DATETIME
DECLARE change_cursor CURSOR FOR
SELECT Operation, CustomerID, CustomerName, City, ChangeDate
FROM @Changes
ORDER BY ChangeLSN
OPEN change_cursor
FETCH NEXT FROM change_cursor INTO @Op, @CustID, @Name, @City, @ChangeDate
WHILE @@FETCH_STATUS = 0
BEGIN
IF @Op IN (3,4) -- Update