Hi!
This blog will outline some high level aspects of implementing service broker, this is targeted at people who are already familiar with the basics of service broker, if you need the basics, check this tutorial:
http://msdn.microsoft.com/en-us/library/bb839483.aspx
Scenario:
You have a production system that is processing millions of records a month.
Reporting requires frequent complex reports to run on the metrics and also on the actual data stored.
The actual data payload is stored as XML data type. This means any reporting metrics on data payload must be done on a fact table of some sort.
Data must be archived for 5 years. (This will be around 500 million records)
Objectives:
Decouple the reporting from the production system
Ensure up to date synchronization from production to the reporting server
Running reports will not affect the production system
Solution:
- We will use service broker to define ONE message contract for all tables that need to be synced. Synchronization will be activated by a job which activates the SQL service broker process
- A message type per table is created
- Synchronization will work in batch mode, supporting running of 40 to 50 simultaneous SQL processes (Optimal for a quad core process = 48)
- After synchronization of the main transaction tables a job on the reporting server will update the FACTS table.
- The initial queue on the sender side will have a PROC activation configured on the queue. When the SQL job runs it will activate this job.
- We will have a service to Trigger the archiving process
- We will create a Route to the destination system
- Up to you how to use the Service Broker security JIn this sample I keep it simple, no encryption. You can use Asymmetric Encryption/security etc, not in scope of this article! In this example I USE A SQL USER MYUser and User, notice the service and route user names are swapped on either server J
The fact table is used to extract information from the XML data. This allows the data to be indexed and reports to run smooth.
Do you have a transaction database and need a custom reporting solution, without the complexity of log shipping or replication? Then look no further, SQL Service broker is the ideal tool to synchronize two databases, based on simple tables.
Source SQL Server
On this server you create a Service Broker Database to activate sending the data contracts payload. So for example you can create a database called SSBSender, this will manage the starting and spawning of processes, queues etc
Contract:
USE
[SSBSender]
GO
CREATE CONTRACT [ArchivingTables] AUTHORIZATION [dbo] ([CONF_AuditTrail] SENT BY ANY,
[Table_Error] SENT BY ANY,
[Table_Record] SENT BY ANY,
[Table_RecordData] SENT BY ANY,
[Table_Transaction] SENT BY ANY)
MessageTypes:
USE [SSBSender]
GO
CREATE MESSAGE TYPE [Table_Error] AUTHORIZATION [dbo] VALIDATION = WELL_FORMED_XML
CREATE MESSAGE TYPE [Table_Record] AUTHORIZATION [dbo] VALIDATION = WELL_FORMED_XML
CREATE MESSAGE TYPE [Table_RecordData] AUTHORIZATION [dbo] VALIDATION = WELL_FORMED_XML
CREATE MESSAGE TYPE [Table_Transaction] AUTHORIZATION [dbo] VALIDATION = WELL_FORMED_XML
Sender Queue:
CREATE QUEUE [dbo].[InitialQueue] WITH STATUS = ON , RETENTION = OFF , ACTIVATION ( STATUS = ON , PROCEDURE_NAME = [dbo].[SSB_ReportTransmission] , MAX_QUEUE_READERS = 20 , EXECUTE AS OWNER ) ON [PRIMARY]
Service:
CREATE SERVICE [TriggerArchiving] AUTHORIZATION [MyUser] ON QUEUE [dbo].[InitialQueue] ([ArchivingTables])
Route
CREATE ROUTE [SSBReceiver] AUTHORIZATION [dbo] WITH SERVICE_NAME = N’ReportingReceivingService’ , ADDRESS = N’TCP://DestinationServer:4022′
Remote Service Binding
CREATE REMOTE SERVICE BINDING [ReceivingServiceBinding] AUTHORIZATION [dbo] TO SERVICE N’ReportingReceivingService’ WITH USER = [MyUserName] , ANONYMOUS = OFF
Activation Job
Schedule a job to activate the process.
declare @datetime datetime
select @datetime = getdate()
EXEC [SSBSender].[dbo].[Start_Report_Transmission] @batch_size = 2500, @archive_date_limit = @datetime
Sender Procedures
USE [SSBSender]
GO
CREATE PROCEDURE [dbo].[Start_Report_Transmission] @batch_size INT = NULL, @archive_date_limit DATETIME = NULL
AS
DECLARE @xmldata XML
DECLARE @InitDlgHandle UNIQUEIDENTIFIER
SET @batch_size = ISNULL(@batch_size, 100)
SET @archive_date_limit = ISNULL(@archive_date_limit, GETDATE())
–@xmldata contains the batch size for every table
SET @xmldata = N'<batch>’ + CAST(@batch_size AS VARCHAR(30)) + N'</batch><archive_date_limit>’ + CONVERT(NVARCHAR(20), @archive_date_limit, 120)+ N'</archive_date_limit>’
–create a dialog from TriggerArchiving to itself
;BEGIN DIALOG @InitDlgHandle
FROM SERVICE TriggerArchiving
TO SERVICE N’TriggerArchiving’
ON CONTRACT ArchivingTables
WITH ENCRYPTION = OFF;
–send as many messages as the tables that need to be sent to reporting
;SEND ON CONVERSATION @InitDlgHandle MESSAGE TYPE [Table_Error](@xmldata);
;SEND ON CONVERSATION @InitDlgHandle MESSAGE TYPE [Table_Transaction](@xmldata);
;SEND ON CONVERSATION @InitDlgHandle MESSAGE TYPE [Table_Record](@xmldata);
;SEND ON CONVERSATION @InitDlgHandle MESSAGE TYPE [Table_RecordData](@xmldata);
END CONVERSATION @InitDlgHandle;
The procesure above will TRIGEER the Queue activation proc below!!!
USE [SSBSender]
GO
/****** Object: StoredProcedure [dbo].[SSB_ReportTransmission] Script Date: 03/10/2010 11:08:23 ******/
SET ANSI_NULLS ON
GO
SET QUOTED_IDENTIFIER ON
GO
CREATE PROCEDURE [dbo].[SSB_ReportTransmission]
AS
DECLARE @batch INT;
DECLARE @archive_date_limit DATETIME;
DECLARE @RecvReqDlgHandle UNIQUEIDENTIFIER;
DECLARE @RecvReqMsg XML;
DECLARE @RecvReqMsgName sysname;
DECLARE @ReportTransmissionHandle UNIQUEIDENTIFIER;
DECLARE @list XML;
DECLARE @far_service NVARCHAR(256)
WHILE 1=1
BEGIN
SELECT @RecvReqDlgHandle = NULL, @RecvReqMsg = NULL, @RecvReqMsgName = NULL, @far_service = NULL
WAITFOR
(
RECEIVE TOP(1)
@RecvReqDlgHandle = conversation_handle,
@RecvReqMsg = message_body,
@RecvReqMsgName = message_type_name
FROM InitialQueue
), TIMEOUT 10000
IF @RecvReqMsgName = N’http://schemas.microsoft.com/SQL/ServiceBroker/EndDialog’
OR @RecvReqMsgName = N’http://schemas.microsoft.com/SQL/ServiceBroker/Error’
BEGIN
IF @RecvReqDlgHandle IS NOT NULL
BEGIN
END CONVERSATION @RecvReqDlgHandle;
END
RETURN 0
END
–at this point there is a message.
–check where the message is coming from (the far service)
SELECT @far_service = far_service
FROM sys.conversation_endpoints
WHERE conversation_handle = @RecvReqDlgHandle
–based on the source of the message perform different actions
IF @far_service = N’ReportingReceivingService’ –that means this is an answer from the reporting server
BEGIN
–set the status of records
EXEC dbo.SetReportingFlag @tabletype = @RecvReqMsgName, @list = @RecvReqMsg
CONTINUE –to receiving loop
END
–at this point the message came from a local call (triggering repeatable transmission)
–get the batch size
SET @batch = @RecvReqMsg.value(‘(batch/text())[1]’, ‘INT’)
SET @archive_date_limit = @RecvReqMsg.value(‘(archive_date_limit/text())[1]’, ‘DATETIME’)
IF ISNULL(@batch, 0) = 0 OR @archive_date_limit IS NULL
BEGIN
RETURN 0
END
;BEGIN DIALOG @ReportTransmissionHandle
FROM SERVICE TriggerArchiving
TO SERVICE N’ReportingReceivingService’
ON CONTRACT ArchivingTables
WITH ENCRYPTION = OFF;
WHILE 1=1
BEGIN
SET @list = NULL
–call procedure that updates the appropriate table
–and returns a list of ids that need to be transferred
EXEC dbo.GetArchiveIdsforReporting @tabletype = @RecvReqMsgName, @batch = @batch, @archive_date_limit = @archive_date_limit, @list = @list OUTPUT
–break the loop when list is null (no records to be transferred)
IF @list IS NULL
BEGIN
BREAK
END
–transmit the list, with the initially received message type
;SEND ON CONVERSATION @ReportTransmissionHandle MESSAGE TYPE @RecvReqMsgName(@list);
END
–end the conversation when no more records are to be transmitted
END CONVERSATION @ReportTransmissionHandle;
–break receiving loop
BREAK
END –of receiving looping
RETURN 0
USE [SSBSender]
GO
/****** Object: StoredProcedure [dbo].[GetArchiveIdsforReporting] Script Date: 03/10/2010 11:09:22 ******/
SET ANSI_NULLS ON
GO
SET QUOTED_IDENTIFIER ON
GO
CREATE PROCEDURE [dbo].[GetArchiveIdsforReporting] @tabletype SYSNAME = NULL, @batch INT = NULL, @archive_date_limit DATETIME = NULL, @list XML = NULL OUTPUT
AS
–first NULLify the output variable
–this is important for always returning the right value (when output param is NULL)
–to the caller
SET @list = NULL
IF ISNULL(@tabletype, ”) = ” OR ISNULL(@batch, 0) = 0 OR @archive_date_limit IS NULL
BEGIN
RETURN
END
DECLARE @tmp TABLE (archive_id BIGINT, tmpid INT IDENTITY(1,1) PRIMARY KEY CLUSTERED (archive_id, tmpid))
–trigger update of table and sending of records
–based on the message type
IF @tabletype = ‘Table_Error’
BEGIN
UPDATE TOP (@batch) MyTransactionDatabase.dbo.tfe_TableError
SET sent_to_report = 1
OUTPUT inserted.tfe_id INTO @tmp(archive_id)
WHERE sent_to_report = 0
AND archive_date <= @archive_date_limit
END
ELSE IF @tabletype = ‘Table_Transaction’
BEGIN
UPDATE TOP (@batch) MyTransactionDatabase.dbo.tft_TableTransaction
SET sent_to_report = 1
OUTPUT inserted.tft_id INTO @tmp(archive_id)
WHERE sent_to_report = 0
AND archive_date <= @archive_date_limit
END
ELSE IF @tabletype = ‘Table_Record’
BEGIN
UPDATE TOP (@batch) MyTransactionDatabase.dbo.tfr_TableRecord
SET sent_to_report = 1
OUTPUT inserted.tfr_id INTO @tmp(archive_id)
WHERE sent_to_report = 0
AND archive_date <= @archive_date_limit
END
ELSE IF @tabletype = ‘Table_RecordData’
BEGIN
UPDATE TOP (@batch) MyTransactionDatabase.dbo.trd_TableRecordData
SET sent_to_report = 1
OUTPUT inserted.trd_id INTO @tmp(archive_id)
WHERE sent_to_report = 0
AND archive_date <= @archive_date_limit
END
IF NOT EXISTS(SELECT * FROM @tmp)
BEGIN
RETURN
END
SET @list = NULL
–create the list of tbl_id’s
SET @list = (SELECT TOP 100 PERCENT archive_id AS id
FROM @tmp
ORDER BY archive_id
FOR XML PATH(”)
)
USE [SSBSender]
GO
/****** Object: StoredProcedure [dbo].[SetReportingFlag] Script Date: 03/10/2010 11:11:55 ******/
SET ANSI_NULLS ON
GO
SET QUOTED_IDENTIFIER ON
GO
CREATE PROCEDURE [dbo].[SetReportingFlag] @tabletype sysname = NULL, @list XML = NULL
AS
DECLARE @TransferStatus TINYINT
IF @tabletype IS NULL OR @list IS NULL
BEGIN
RETURN 0
END
–get the transmission flag value
SET @TransferStatus = @list.value(‘(TransferStatus/text())[1]’, ‘TINYINT’)
IF @TransferStatus IS NULL
BEGIN
RETURN 0
END
–shred the @list and get id values
DECLARE @tmp TABLE (archive_id BIGINT, tmpid INT IDENTITY(1,1) PRIMARY KEY CLUSTERED (archive_id, tmpid))
–shred the @list of values
INSERT INTO @tmp(archive_id)
SELECT a.col.value(‘text()[1]’, ‘BIGINT’)
FROM @list.nodes(‘id’) a(col)
IF NOT EXISTS(SELECT TOP 1 * FROM @tmp)
BEGIN
RETURN 0
END
–update proper table with @TransferStatus
IF @tabletype = ‘Table_Error’
BEGIN
UPDATE archive_tbl
SET sent_to_report = @TransferStatus
FROM MyTransactionDatabase.dbo.tfe_TableError archive_tbl
JOIN @tmp tmp ON archive_tbl.tfe_id = tmp.archive_id
END
ELSE IF @tabletype = ‘Table_Transaction’
BEGIN
UPDATE archive_tbl
SET sent_to_report = @TransferStatus
FROM MyTransactionDatabase.dbo.tft_TaleTransaction archive_tbl
JOIN @tmp tmp ON archive_tbl.tft_id = tmp.archive_id
END
ELSE IF @tabletype = ‘Table_Record’
BEGIN
UPDATE archive_tbl
SET sent_to_report = @TransferStatus
FROM MyTransactionDatabase.dbo.tfr_tableRecord archive_tbl
JOIN @tmp tmp ON archive_tbl.tfr_id = tmp.archive_id
END
ELSE IF @tabletype = ‘Table_RecordData’
blah blah blah
RETURN 0
Destination SQL Server
This is the server that will receive the data and process the messages in the queue. The database here to manage synchronization will be SSBReceiver, this will manage the receiving queuing and parsing the message payload to the receiver procedure.
On the receiving database, you will have the SAME contract and message types.
Queue
CREATE QUEUE [dbo].[ReceivingQueue] WITH STATUS = ON , RETENTION = OFF , ACTIVATION ( STATUS = ON , PROCEDURE_NAME = [dbo].[SSBReceiveMessage] , MAX_QUEUE_READERS = 40 , EXECUTE AS OWNER ) ON [PRIMARY]
Service
CREATE SERVICE [ReportingReceivingService] AUTHORIZATION [User] ON QUEUE [dbo].[ReceivingQueue] ([ArchivingTables])
Route
CREATE ROUTE [TriggerArchiving] AUTHORIZATION [dbo] WITH SERVICE_NAME = N’TriggerArchiving’ , ADDRESS = N’TCP://SourceServer:4022′
Remote Service Binding
CREATE REMOTE SERVICE BINDING [TriggerArchivingServiceBinding] AUTHORIZATION [dbo] TO SERVICE N’TriggerArchiving’ WITH USER = [MyUserName] , ANONYMOUS = OFF
Now all you need to do is get the stored procedure that is activated on the receiving queue to manage the message payload JEasy as that!
USE [SSBReceiver]
GO
/****** Object: StoredProcedure [dbo].[SSBReceiveMessage] Script Date: 03/10/2010 11:20:17 ******/
SET ANSI_NULLS ON
GO
SET QUOTED_IDENTIFIER ON
GO
CREATE PROCEDURE [dbo].[SSBReceiveMessage]
AS
print ‘SSBReceiveMessage started’
DECLARE @RecvReqDlgHandle UNIQUEIDENTIFIER;
DECLARE @RecvReqMsg XML;
DECLARE @RecvReqMsgName SYSNAME;
DECLARE @ReportAnswerHandle UNIQUEIDENTIFIER;
–begin an answering dialogue with the TriggerArchiving service
;BEGIN DIALOG @ReportAnswerHandle
FROM SERVICE ReportingReceivingService
TO SERVICE N’TriggerArchiving’
ON CONTRACT ArchivingTables
WITH ENCRYPTION = OFF;
WHILE 1=1
BEGIN
SELECT @RecvReqDlgHandle = NULL, @RecvReqMsg = NULL, @RecvReqMsgName = NULL;
WAITFOR
(
RECEIVE TOP(1)
@RecvReqDlgHandle = conversation_handle,
@RecvReqMsg = message_body,
@RecvReqMsgName = message_type_name
FROM [ReceivingQueue]
), TIMEOUT 10000
IF @RecvReqMsgName = N’http://schemas.microsoft.com/SQL/ServiceBroker/EndDialog’
OR @RecvReqMsgName = N’http://schemas.microsoft.com/SQL/ServiceBroker/Error’
BEGIN
IF @RecvReqDlgHandle IS NOT NULL
BEGIN
END CONVERSATION @RecvReqDlgHandle;
END
BREAK
END
–in case the message is null, break looping
IF @RecvReqMsg IS NULL
BEGIN
BREAK
END
BEGIN TRY
–execute procedure that pulls archived records
EXEC dbo.PullArchiveRecords @tabletype = @RecvReqMsgName, @list = @RecvReqMsg
–apply Success Status element in @RecvReqMsg
SET @RecvReqMsg.modify(‘insert element TransferStatus {2} as first into .’)
END TRY
BEGIN CATCH
–at this point an error occured
print ‘Error occured during PullArchiveRecords: ‘ + ERROR_MESSAGE()
–apply Error Status element in @RecvReqMsg
SET @RecvReqMsg.modify(‘insert element TransferStatus {3} as first into .’)
END CATCH
–send the action answer to TriggerArchiving
;SEND ON CONVERSATION @ReportAnswerHandle MESSAGE TYPE @RecvReqMsgName(@RecvReqMsg);
END
–end conversation with TriggerArchiving service
END CONVERSATION @ReportAnswerHandle
RETURN 0
From here, you CREATE your own logic on the calling proc from the above proc to transfer the batches to the reporting/Archive tables.
USE [SSBReceiver]
GO
SET ANSI_NULLS ON
GO
SET QUOTED_IDENTIFIER ON
GO
CREATE PROCEDURE [dbo].[PullArchiveRecords] @tabletype SYSNAME = NULL, @list XML = NULL
AS
IF @tabletype IS NULL OR @list IS NULL
BEGIN
RETURN 0
END
DECLARE @tmp TABLE (archive_id BIGINT, tmpid INT IDENTITY(1,1) PRIMARY KEY CLUSTERED (archive_id, tmpid))
–shred the @list of values
INSERT INTO @tmp(archive_id)
SELECT a.col.value(‘text()[1]’, ‘BIGINT’)
FROM @list.nodes(‘id’) a(col)
IF NOT EXISTS(SELECT TOP 1 * FROM @tmp)
BEGIN
RETURN 0
END
–pull records from Archiving server
–based on @tabletype
BEGIN TRY
IF @tabletype = ‘Table_Error’
BEGIN
–PUT LOGIC HERE to insert from payload joining on @tmp
END
ELSE IF @tabletype = ‘Table_Transaction’
BEGIN
–PUT LOGIC HERE to insert from payload joining on @tmp
END
ELSE IF @tabletype = ‘Table_Record’
BEGIN
–PUT LOGIC HERE to insert from payload joining on @tmp
END
ELSE IF @tabletype = ‘Table_RecordData’
BEGIN
–PUT LOGIC HERE to insert from payload joining on @tmp
END
END TRY
BEGIN CATCH
INSERT INTO dbo.myerrors(errormessage)
values(ERROR_MESSAGE())
END CATCH
Reporting Server Facts Job
You can then create a FACT Table job to update the facts table with data that was just imported. You can use a logical flag to ensure data is flagged in the procedure above, where you implement your logic when inserting the new data. The fact job can then use a flag value to update facts, similar to a CUBE on an OLAP server.
Conclusion
You can develop some super fast and stream lime synchronization mechanisms with service broker, transferring millions of records is a breeze, since batch sizes are kept small and we spawn 48-50 parrallel batching processes at once into the queue, you ensure your Transaction log checkpoints are efficient, especially on high processing load days. Much more efficient that Replication and Log Shipping, and MUCH easier to support. This means, we can always keep the queue size at a max of +- 50 🙂 Easy to manage on large transaction days.
I have transferred around 300 million records and not had a hiccup since.
Romiko Derbynew 
You must be logged in to post a comment.