Triển khai tải gia tăng bằng cách sử dụng Thay đổi tính năng thu thập dữ liệu trong SQL Server

Bài viết này sẽ rất thú vị đối với những người thường xuyên phải xử lý tích hợp dữ liệu.

Giới thiệu

Giả sử rằng có một cơ sở dữ liệu nơi người dùng luôn sửa đổi dữ liệu (cập nhật hoặc loại bỏ). Có lẽ, cơ sở dữ liệu này được sử dụng bởi một ứng dụng lớn không cho phép sửa đổi cấu trúc bảng. Nhiệm vụ là thỉnh thoảng tải dữ liệu từ cơ sở dữ liệu này sang cơ sở dữ liệu khác trên một máy chủ khác. Cách đơn giản nhất để giải quyết vấn đề là tải dữ liệu mới từ cơ sở dữ liệu nguồn sang cơ sở dữ liệu đích với việc dọn dẹp sơ bộ cơ sở dữ liệu đích. Bạn có thể sử dụng phương pháp này miễn là thời gian tải dữ liệu có thể chấp nhận được và không vượt quá thời hạn đặt trước. Điều gì sẽ xảy ra nếu mất vài ngày để tải dữ liệu? Ngoài ra, các kênh liên lạc không ổn định dẫn đến tình trạng tải dữ liệu dừng và khởi động lại. Nếu bạn gặp phải những trở ngại này, tôi khuyên bạn nên xem xét một trong các thuật toán ‘tải lại dữ liệu’. Điều đó có nghĩa là chỉ xảy ra các sửa đổi dữ liệu kể từ khi tải mới nhất được tải.

CDC

Trong SQL Server 2008, Microsoft đã giới thiệu một cơ chế theo dõi dữ liệu được gọi là Change Data Capture (CDC). Nói chung, mục đích của cơ chế này là cho phép CDC cho bất kỳ bảng cơ sở dữ liệu nào sẽ tạo ra một bảng hệ thống trong cùng một cơ sở dữ liệu với tên tương tự như bảng gốc (lược đồ sẽ như sau:'cdc' làm tiền tố cộng với tên lược đồ cũ cộng với ”_” và kết thúc là “_CT”. Ví dụ:bảng gốc là dbo. Ví dụ, sau đó bảng hệ thống sẽ được gọi là cdc.dbo_Example_CT). Nó sẽ lưu trữ tất cả dữ liệu đã được sửa đổi.

Trên thực tế, để tìm hiểu sâu hơn về CDC, hãy xem xét ví dụ. Nhưng trước tiên, hãy đảm bảo rằng SQL Agent sử dụng CDC hoạt động trên phiên bản thử nghiệm SQL Server.

Ngoài ra, chúng tôi sẽ xem xét một tập lệnh tạo cơ sở dữ liệu và bảng kiểm tra, điền dữ liệu vào bảng này và bật CDC cho bảng này.

Để hiểu và đơn giản hóa tác vụ, chúng tôi sẽ sử dụng một phiên bản SQL Server mà không phân phối cơ sở dữ liệu nguồn và cơ sở dữ liệu đích cho các máy chủ khác nhau.

 sử dụng mastergo-- tạo cơ sở dữ liệu nguồn nếu không tồn tại (chọn * từ sys.databases trong đó name ='db_src_cdc') tạo cơ sở dữ liệu db_src_cdcgouse db_src_cdcgo-- bật CDC nếu nó bị tắt nếu không tồn tại (chọn * từ sys.databases trong đó tên =db_name () and is_cdc_enabled =1) execute sys.sp_cdc_enable_dbgo-- tạo vai trò cho các bảng không tồn tại CDCif (chọn * từ sys.sysusers trong đó name ='CDC_Reader' và Issqlrole =1) tạo vai trò CDC_Readergo-- tạo một tableif object_id ('dbo.Example', 'U') là null tạo bảng dbo.Ví dụ (ID int ràng buộc định danh PK_Example khóa chính, Title varchar (200) không phải null) go-- điền giá trị dbo của tableinsert. Ví dụ (Title) ( 'Một'), ('Hai'), ('Ba'), ('Bốn'), ('Năm'); go-- bật CDC cho bảng nếu không tồn tại (chọn * từ sys.tables trong đó is_tracked_by_cdc =1 và name ='Ví dụ') thi hành sys.sp_cdc_enable_table @source_schema ='dbo', @source_name ='Ví dụ', @role_name ='CDC_Reader'go-- điền một số dữ liệu vào bảng. Chúng tôi sẽ thay đổi hoặc xóa cái gì đó cập nhật dbo.Exampleset Tiêu đề =đảo ngược (Tiêu đề) trong đó ID trong (2,3,4); xóa khỏi dbo.Ví dụ trong đó ID trong (1,2); đặt ID_insert dbo. Ví dụ trên; chèn dbo. Giá trị ví dụ (ID, Tiêu đề) (1, 'Một'), (6, 'Sáu'); đặt ID_insert dbo. Ví dụ tắt; go

Bây giờ, hãy xem những gì chúng ta có sau khi thực thi tập lệnh này trong bảng dbo.Example và cdc.dbo_Example_CT (cần lưu ý rằng CDC là không đồng bộ. Dữ liệu được điền vào các bảng nơi theo dõi thay đổi được lưu trữ sau một khoảng thời gian nhất định ).

 select * from dbo.Example;

 ID Title ---------------------- 1 One 3 eerhT 4 ruoF 5 Five 6 Six

 select row_number () over (phân vùng theo thứ tự ID bởi __ $ start_lsn desc, __ $ seqval desc) là __ $ rn, * từ cdc.dbo_Example_CT;

 __ $ rn __ $ start_lsn __ $ end_lsn __ $ seqval __ $ hoạt động __ $ update_mask Tiêu đề ID -------------------------- - ----------- ---------------------------- ------------ --- ----------- 1 0x0000003A000000580005 NULL 0x0000003A000000580003 2 0x03 1 One 2 0x0000003A000000560006 NULL 0x0000003A000000560002 1 0x03 1 One 1 0x0000003A ow0000560006 NULL 1 0x0000003 2 0x0000003A000000540005 NULL 0x0000003A000000540002 3 0x02 2 Two 3 0x0000003A000000540005 NULL 0x0000003A000000540002 4 0x02 2 owT 1 0x0000003A000000540005 NULL 0x0000003A000000540003 3 0x02 3 Three 2 0x0000003A000000540005 NULL 0x0000003A000000540003 4 0x02 3 eerhT 1 0x0000003A000000540005 NULL 0x0000003A000000540004 3 0x02 4 Bốn 2 0x0000003A000000540005 NULL 0x0000003A000000540004 4 0x02 4 ruoF 1 0x0000003A000000580005 NULL 0x0000003A000000580004 2 0x03

Xem xét chi tiết cấu trúc bảng mà theo dõi thay đổi được lưu trữ. Trường __ $ start_lsn và __ $ seqval lần lượt là LSN (số thứ tự nhật ký trong cơ sở dữ liệu) và số giao dịch trong giao dịch. Có một thuộc tính quan trọng trong các trường này, đó là, chúng ta có thể chắc chắn rằng bản ghi có LSN cao hơn sẽ được thực hiện sau này. Do thuộc tính này, chúng tôi có thể dễ dàng nhận được trạng thái mới nhất của mỗi bản ghi trong truy vấn, lọc lựa chọn của chúng tôi theo điều kiện - trong đó __ $ rn =1.

Trường hoạt động __ $ chứa mã giao dịch:

1 - bản ghi đã bị xóa
2 - bản ghi được chèn
3, 4 - bản ghi được cập nhật. Dữ liệu cũ trước khi cập nhật là 3, dữ liệu mới là 4.

Ngoài các trường dịch vụ có tiền tố «__ $», các trường của bảng gốc hoàn toàn trùng lặp. Thông tin này đủ để chúng tôi tiếp tục tải gia tăng.

Thiết lập cơ sở dữ liệu để tải dữ liệu

Tạo một bảng trong cơ sở dữ liệu mục tiêu thử nghiệm của chúng tôi, trong đó dữ liệu sẽ được tải, cũng như một bảng bổ sung để lưu trữ dữ liệu về nhật ký tải.

 sử dụng mastergo-- tạo cơ sở dữ liệu đích nếu không tồn tại (chọn * từ sys.databases trong đó name ='db_dst_cdc') tạo cơ sở dữ liệu db_dst_cdcgouse db_dst_cdcgo-- tạo tableif object_id ('dbo.Example', 'U') là null tạo bảng dbo. Ví dụ (khóa chính ID int R ràng buộc PK_Example, Tiêu đề varchar (200) không phải null) go-- tạo bảng để lưu trữ logif object_id ('dbo.log_cdc', 'U') là rỗng tạo bảng dbo .log_cdc (table_name nvarchar (512) không phải null, dt datetime không null default getdate (), lsn binary (10) not null default (0x0), khóa chính pk_log_cdc hạn chế (table_name, dt desc)) đi

Tôi muốn thu hút sự chú ý của bạn đến các trường của bảng LOG_CDC:

TABLE_NAME lưu trữ thông tin về bảng đã được tải (có thể tải một số bảng trong tương lai, từ các cơ sở dữ liệu khác nhau hoặc thậm chí từ các máy chủ khác nhau; định dạng bảng là 'SERVER_NAME.DB_NAME.SCHEMA_NAME.TABLE_NAME'
DT là trường ngày và giờ tải, là trường tùy chọn cho tải tăng dần. Tuy nhiên, nó sẽ hữu ích cho việc kiểm tra tải.
LSN - sau khi một bảng được tải, chúng tôi cần lưu trữ thông tin về nơi bắt đầu tải tiếp theo, nếu được yêu cầu. Theo đó, sau mỗi lần tải, chúng tôi thêm __ $ start_lsn mới nhất (tối đa) vào cột này.

Thuật toán tải dữ liệu

Như đã mô tả ở trên, bằng cách sử dụng truy vấn, chúng ta có thể có được trạng thái mới nhất của bảng với sự trợ giúp của các hàm cửa sổ. Nếu chúng ta biết LSN của lần tải gần nhất, trong lần tải tiếp theo, chúng ta có thể lọc từ nguồn tất cả dữ liệu, những thay đổi trong đó cao hơn LSN được lưu trữ, nếu có ít nhất một lần tải hoàn chỉnh trước đó:

 with incr_Example as (select row_number () over (phân vùng theo thứ tự ID bởi __ $ start_lsn desc, __ $ seqval desc) dưới dạng __ $ rn, * từ db_src_cdc.cdc.dbo_Example_CT trong đó __ $ hoạt động <> 3 và __ $ start_lsn> @lsn) chọn * từ incr_Example

Sau đó, chúng tôi có thể lấy tất cả các bản ghi cho tải hoàn chỉnh, nếu LSN tải không được lưu trữ:

with incr_Example as (select row_number () over (phân vùng theo thứ tự ID bởi __ $ start_lsn desc, __ $ seqval desc) dưới dạng __ $ rn, * từ db_src_cdc.cdc.dbo_Example_CT trong đó __ $ hoạt động <> 3 và __ $ start_lsn> @lsn), full_Example as (select * from db_src_cdc.dbo. Ví dụ trong đó @lsn là null) chọn ID, Title, __ $ operationfrom incr_Examplewhere __ $ rn =1union chọn tất cả ID, Title, 2 as __ $ operationfrom full_Example

Do đó, tùy thuộc vào giá trị @LSN, truy vấn này sẽ hiển thị tất cả các thay đổi mới nhất (bỏ qua những thay đổi tạm thời) với trạng thái Đã xóa hoặc không, hoặc tất cả dữ liệu từ bảng gốc, thêm trạng thái 2 (bản ghi mới) - trường này chỉ được sử dụng để hợp nhất hai lựa chọn. Với truy vấn này, chúng tôi có thể dễ dàng thực hiện tải đầy đủ hoặc tải lại bằng lệnh MERGE (bắt đầu với phiên bản SQL 2008).

Để tránh tắc nghẽn có thể tạo ra các quy trình thay thế và tải dữ liệu phù hợp từ các bảng khác nhau (trong tương lai, chúng tôi sẽ tải một số bảng và có thể có các mối quan hệ giữa chúng), tôi khuyên bạn nên sử dụng ảnh chụp nhanh DB trên cơ sở dữ liệu nguồn ( một tính năng khác của SQL 2008).

Nội dung đầy đủ của tải như sau:

[expand title =”Mã”]

 / * Thuật toán tải dữ liệu * / - tạo cơ sở dữ liệu snapshotif tồn tại (chọn * từ sys.databases nơi name ='db_src_cdc_ss') thả cơ sở dữ liệu db_src_cdc_ss; khai báo @query nvarchar (max); select @query =N ' tạo cơ sở dữ liệu db_src_cdc_ss trên (name =N '' '+ name +' '', filename =N '' '+ [tên tệp] +'. ss '') dưới dạng ảnh chụp nhanh của db_src_cdc'from db_src_cdc.sys.sysfiles where groupid =1; exe một bảng trước khi hoàn thành loadif @lsn là null, cắt ngắn bảng db_dst_cdc.dbo.Example; - load processwith incr_Example as (select row_number () over (phân vùng theo thứ tự ID bởi __ $ start_lsn desc, __ $ seqval desc) as __ $ rn , * từ db_src_cdc_ss.cdc.dbo_Example_CT trong đó __ $ operation <> 3 và __ $ start_lsn> @lsn), full_Example as (select * from db_src_cdc_ss.dbo. Ví dụ trong đó @lsn là null), cte_Example as (select ID, Tiêu đề, __ $ hoạt động từ incr_Example trong đó __ $ rn =1 liên hiệp tất cả chọn ID, Tiêu đề, 2 là __ $ hoạt động từ full_Example) hợp nhất db_dst_cdc.dbo. Ví dụ như trg bằng cách sử dụng cte_Example as src trên trg.ID =src.IDkhi khớp và __ $ operation =1 sau đó xóa khi khớp và __ $ hoạt động <> 1 sau đó cập nhật tập hợp trg.Title =src.Title Khi không khớp theo mục tiêu và __ $ hoạt động <> 1 thì chèn các giá trị (ID, Title) (src.ID, src .Title); - đánh dấu phần cuối của quá trình tải và giá trị LSNinsert db_dst_cdc.dbo.log_cdc (table_name, lsn) mới nhất ('localhost.db_src_cdc.dbo.Example', isnull ((select max (__ $ start_lsn) from db_src_cdc_ss.cdc.dbo_Example_CT), 0)) - xóa snapshotif cơ sở dữ liệu tồn tại (chọn * từ sys.databases nơi name ='db_src_cdc_ss') thả cơ sở dữ liệu db_src_cdc_ss

[/ mở rộng]