У меня такое ощущение, что этот ответ может быть излишним, но я думаю, что все это полезно при обработке передачи большого объема ETL с Oracle на SQL Сервер.
В этом примере я собираюсь потянуть детали счета. Это записи отдельных позиций для счетов-фактур (т.е. общих продаж) для крупной компании. Таблица, из которой я извлекаю в Oracle, содержит около 1 миллиарда строк, и мне нужно около 300 миллионов записей каждый раз. После того, как я их перетяну, я сравниваю их с тем, что у меня есть на SQL Server, и затем делаю необходимые ОБНОВЛЕНИЯ, ВСТАВКИ или УДАЛЕНИЯ. И я делаю это несколько раз в день, и теперь это занимает всего около 20 минут на весь процесс. Но в начале весь процесс занял несколько часов.
Чтобы извлечь необработанные данные из Oracle, я использую довольно простой пакет, который генерирует запрос для использования в Oracle, обрезает мой SQL сервер промежуточная таблица импорта (у меня есть еще несколько промежуточных таблиц позже), извлекает данные из Oracle, а затем вызывает pro c, который выполняет объединение (не фактическое MERGE, как мы увидим). Поток управления выглядит следующим образом:
Первым важным шагом является максимально возможное ограничение данных на стороне Oracle. Я не хочу тянуть все 1 миллиард строк каждый раз, когда запускается этот пакет. Поэтому я ограничиваю записи только теми записями, где Invoice Date больше или равно 1 января два года go. Но поскольку форматы даты между SQL Server и Oracle не очень совместимы (лучший метод, который я нашел для преобразования обоих в один и тот же строковый формат), я должен сделать некоторое форматирование, и потому что я хочу дату диапазон всегда должен быть корректным, исходя из дня, в который я работаю, мне нужно немного подсчитать встроенную дату.
SELECT
CAST("Data Source Code" AS VARCHAR2(3)) AS "DataSourceCode"
,CAST("Order#" AS VARCHAR2(11)) AS "OrderNum"
,CAST("Invoice#" AS VARCHAR2(15)) AS "InvoiceNum"
,CAST("Item#" AS VARCHAR2(10)) AS "ItemNumber"
,CAST("Order Line Type" AS VARCHAR2(10)) AS "OrderLineType"
,CAST("Order Status" AS VARCHAR2(1)) AS "OrderStatus"
,"Order Date Time" AS "OrderDate"
,CAST("Fiscal Invoice Period" AS VARCHAR2(6)) AS "FiscalInvoicePeriod"
,"Invoice Date" AS "InvoiceDate"
,CAST("Ship To Cust#" AS VARCHAR2(8)) AS "ShipToCustNum"
,CAST("Billing Account #" AS VARCHAR2(8)) AS "BillingAccountNumber"
,CAST("Sales Branch Number" AS VARCHAR2(4)) AS "SalesBranchNumber"
,CAST("Price Branch Number" AS VARCHAR2(4)) AS "PriceBranchNumber"
,CAST("Ship Branch Number" AS VARCHAR2(4)) AS "ShippingBranchNumber"
,"Sold Qty" AS "SoldQty"
,"Unit Price" AS "UnitPrice"
,"Sales Amount" AS "SalesAmount"
,"Handling Amount" AS "HandlingAmount"
,"Freight Amount" AS "FreightAmount"
,"Unit Cogs Amount" AS "UnitCogsAmount"
,"Cogs Amount" AS "CogsAmount"
,"Unit Commcost Amount" AS "UnitCommcostAmount"
,"Commcost Amount" AS "CommcostAmount"
,CAST("GL Period" AS VARCHAR2(6)) AS "GLPeriod"
,"Order Qty" AS "OrderQty"
,"Margin %" AS "MarginPct"
,CONVERT("PO Number",'AL32UTF8','WE8MSWIN1252') AS "PONumber"
,CAST("Branch Id" AS VARCHAR2(4)) AS "BranchId"
,CAST("Outside Salesrep SSO" AS VARCHAR2(20)) AS "OutsideSalesrepSSO"
,CAST("Inside Salesrep SSO" AS VARCHAR2(20)) AS "InsideSalesrepSSO"
,CAST("Order Writer SSO" AS VARCHAR2(20)) AS "OrderWriterSSO"
,"Line Number" AS "LineNumber"
,CAST(UPPER(RAWTOHEX(SYS.DBMS_OBFUSCATION_TOOLKIT.MD5(input_string =>
COALESCE(CAST("Data Source Code" AS VARCHAR2(4)),'') || '|' ||
COALESCE(CAST("Order#" AS VARCHAR2(40)),'') || '|' ||
COALESCE(CAST("Invoice#" AS VARCHAR2(15)),'') || '|' ||
COALESCE(CAST("Item#" AS VARCHAR2(10)),'') || '|' ||
COALESCE(CAST("Order Line Type" AS VARCHAR2(6)),'') || '|' ||
COALESCE(CAST("Order Status" AS VARCHAR2(3)),'') || '|' ||
COALESCE(CAST("Order Date" AS VARCHAR2(19)),'') || '|' ||
COALESCE(CAST("Invoice Date" AS VARCHAR2(19)),'') || '|' ||
COALESCE(CAST("Ship To Cust#" AS VARCHAR2(10)),'') || '|' ||
COALESCE(CAST("Billing Account #" AS VARCHAR2(10)),'') || '|' ||
COALESCE(CAST("Sales Branch Number" AS VARCHAR2(4)),'') || '|' ||
COALESCE(CAST("Price Branch Number" AS VARCHAR2(4)),'') || '|' ||
COALESCE(CAST("Ship Branch Number" AS VARCHAR2(4)),'') || '|' ||
COALESCE(CAST("Sold Qty" AS VARCHAR2(20)),'') || '|' ||
COALESCE(CAST(CAST("Unit Price" AS NUMBER(18,2)) AS VARCHAR2(20)),'') || '|' ||
COALESCE(CAST(CAST("Sales Amount" AS NUMBER(18,2)) AS VARCHAR2(20)),'') || '|' ||
COALESCE(CAST(CAST("Handling Amount" AS NUMBER(18,2)) AS VARCHAR2(20)),'') || '|' ||
COALESCE(CAST(CAST("Freight Amount" AS NUMBER(18,2)) AS VARCHAR2(20)),'') || '|' ||
COALESCE(CAST(TO_DATE('01-' || "Fiscal Invoice Period", 'DD-MON-YY') AS VARCHAR2(19)),'') || '|' ||
COALESCE(CAST(TO_DATE('01-' || "GL Period", 'DD-MON-YY') AS VARCHAR2(19)),'') || '|' ||
COALESCE(CAST("Order Qty" AS VARCHAR2(20)),'') || '|' ||
COALESCE(CAST(CAST("Commcost Amount" AS NUMBER(18,2)) AS VARCHAR2(20)),'') || '|' ||
COALESCE(CAST("Order Writer SSO" AS VARCHAR2(36)),'') || '|' ||
COALESCE(CAST("Line Number" AS VARCHAR2(10)),'')
))) AS VARCHAR2(32)) AS "HashVal"
FROM MY_SCHEMA.MY_INVOICE_TABLE
WHERE "Invoice Date" >= TO_DATE( CONVERT(VARCHAR(10), DATEADD(YEAR,-2,DATEADD(MONTH,1-(DATEPART(MONTH,DATEADD(DAY,1-(DATEPART(DAY,GETDATE())),GETDATE()))),DATEADD(DAY,1-(DATEPART(DAY,GETDATE())),GETDATE()))), 111), 'yyyy/mm/dd')
Вы также заметите, что я вычисляю MD5 Ha sh ценность также. Эту технику я выбрал у очень классного Энди Леонарда . Функция Ha sh принимает строку и применяет одностороннюю криптографическую кодировку строки c на основе выбранного вами алгоритма. MD5 - самый сильный вариант для DBMS_OBFUSCATION_TOOLKIT, который у меня есть с моими ограниченными разрешениями на сервере Oracle. Сервер SQL также может генерировать те же значения Ha sh в MD5, что очень полезно.
Таким образом, общая идея заключается в том, что вы можете преобразовать все поля записи, которые вам нужны о отслеживании в строки и объединении их вместе, примените алгоритм хеширования, чтобы получить HashVal, затем присоедините исходную запись к записи назначения и сравните HashVals. Если HashVals не совпадают, вы знаете, что исходная запись изменилась, и вам нужно обновить запись. Если HashVal не изменился, то исходная запись, вероятно, не изменилась. Я говорю «вероятно», потому что даже с разделителями, вставленными в объединенную строку (чтобы мы могли различать guish между '1a2b' + '3c' + '4d' и '1a' + '2b3c' + '4d'; без разделителя: '1a2b3c4b' против '1a2b3c4b'; с разделителем: '1a2|b3c|4d' против '1a|2b3c|4d') все еще возможно, что MD5 га sh двух разных строк может быть одинаковым.
Но мы хотим динамически сгенерировать этот запрос, поэтому нам нужно вставить его в Execute SQL Task и присвоить его переменной, а затем сопоставить эту переменную с выходным параметром. Строковый запрос затем становится:
SELECT SourceQuery =
'SELECT
CAST("Data Source Code" AS VARCHAR2(3)) AS "DataSourceCode"
,CAST("Order#" AS VARCHAR2(11)) AS "OrderNum"
,CAST("Invoice#" AS VARCHAR2(15)) AS "InvoiceNum"
,CAST("Item#" AS VARCHAR2(10)) AS "ItemNumber"
,CAST("Order Line Type" AS VARCHAR2(10)) AS "OrderLineType"
,CAST("Order Status" AS VARCHAR2(1)) AS "OrderStatus"
,"Order Date Time" AS "OrderDate"
,CAST("Fiscal Invoice Period" AS VARCHAR2(6)) AS "FiscalInvoicePeriod"
,"Invoice Date" AS "InvoiceDate"
,CAST("Ship To Cust#" AS VARCHAR2(8)) AS "ShipToCustNum"
,CAST("Billing Account #" AS VARCHAR2(8)) AS "BillingAccountNumber"
,CAST("Sales Branch Number" AS VARCHAR2(4)) AS "SalesBranchNumber"
,CAST("Price Branch Number" AS VARCHAR2(4)) AS "PriceBranchNumber"
,CAST("Ship Branch Number" AS VARCHAR2(4)) AS "ShippingBranchNumber"
,"Sold Qty" AS "SoldQty"
,"Unit Price" AS "UnitPrice"
,"Sales Amount" AS "SalesAmount"
,"Handling Amount" AS "HandlingAmount"
,"Freight Amount" AS "FreightAmount"
,"Unit Cogs Amount" AS "UnitCogsAmount"
,"Cogs Amount" AS "CogsAmount"
,"Unit Commcost Amount" AS "UnitCommcostAmount"
,"Commcost Amount" AS "CommcostAmount"
,CAST("GL Period" AS VARCHAR2(6)) AS "GLPeriod"
,"Order Qty" AS "OrderQty"
,"Margin %" AS "MarginPct"
,CONVERT("PO Number",''AL32UTF8'',''WE8MSWIN1252'') AS "PONumber"
,CAST("Branch Id" AS VARCHAR2(4)) AS "BranchId"
,CAST("Outside Salesrep SSO" AS VARCHAR2(20)) AS "OutsideSalesrepSSO"
,CAST("Inside Salesrep SSO" AS VARCHAR2(20)) AS "InsideSalesrepSSO"
,CAST("Order Writer SSO" AS VARCHAR2(20)) AS "OrderWriterSSO"
,"Line Number" AS "LineNumber"
,CAST(UPPER(RAWTOHEX(SYS.DBMS_OBFUSCATION_TOOLKIT.MD5(input_string =>
COALESCE(CAST("Data Source Code" AS VARCHAR2(4)),'''') || ''|'' ||
COALESCE(CAST("Order#" AS VARCHAR2(40)),'''') || ''|'' ||
COALESCE(CAST("Invoice#" AS VARCHAR2(15)),'''') || ''|'' ||
COALESCE(CAST("Item#" AS VARCHAR2(10)),'''') || ''|'' ||
COALESCE(CAST("Order Line Type" AS VARCHAR2(6)),'''') || ''|'' ||
COALESCE(CAST("Order Status" AS VARCHAR2(3)),'''') || ''|'' ||
COALESCE(CAST("Order Date" AS VARCHAR2(19)),'''') || ''|'' ||
COALESCE(CAST("Invoice Date" AS VARCHAR2(19)),'''') || ''|'' ||
COALESCE(CAST("Ship To Cust#" AS VARCHAR2(10)),'''') || ''|'' ||
COALESCE(CAST("Billing Account #" AS VARCHAR2(10)),'''') || ''|'' ||
COALESCE(CAST("Sales Branch Number" AS VARCHAR2(4)),'''') || ''|'' ||
COALESCE(CAST("Price Branch Number" AS VARCHAR2(4)),'''') || ''|'' ||
COALESCE(CAST("Ship Branch Number" AS VARCHAR2(4)),'''') || ''|'' ||
COALESCE(CAST("Sold Qty" AS VARCHAR2(20)),'''') || ''|'' ||
COALESCE(CAST(CAST("Unit Price" AS NUMBER(18,2)) AS VARCHAR2(20)),'''') || ''|'' ||
COALESCE(CAST(CAST("Sales Amount" AS NUMBER(18,2)) AS VARCHAR2(20)),'''') || ''|'' ||
COALESCE(CAST(CAST("Handling Amount" AS NUMBER(18,2)) AS VARCHAR2(20)),'''') || ''|'' ||
COALESCE(CAST(CAST("Freight Amount" AS NUMBER(18,2)) AS VARCHAR2(20)),'''') || ''|'' ||
COALESCE(CAST(TO_DATE(''01-'' || "Fiscal Invoice Period", ''DD-MON-YY'') AS VARCHAR2(19)),'''') || ''|'' ||
COALESCE(CAST(TO_DATE(''01-'' || "GL Period", ''DD-MON-YY'') AS VARCHAR2(19)),'''') || ''|'' ||
COALESCE(CAST("Order Qty" AS VARCHAR2(20)),'''') || ''|'' ||
COALESCE(CAST(CAST("Commcost Amount" AS NUMBER(18,2)) AS VARCHAR2(20)),'''') || ''|'' ||
COALESCE(CAST("Order Writer SSO" AS VARCHAR2(36)),'''') || ''|'' ||
COALESCE(CAST("Line Number" AS VARCHAR2(10)),'''')
))) AS VARCHAR2(32)) AS "HashVal"
FROM MY_SCHEMA.MY_INVOICE_TABLE
WHERE "Invoice Date" >= TO_DATE('''+CONVERT(VARCHAR(10), DATEADD(YEAR,-2,DATEADD(MONTH,1-(DATEPART(MONTH,DATEADD(DAY,1-(DATEPART(DAY,GETDATE())),GETDATE()))),DATEADD(DAY,1-(DATEPART(DAY,GETDATE())),GETDATE()))), 111)+''', ''yyyy/mm/dd'')'
В потоке данных у меня есть Oracle Source и OLE DB Destination. Сначала мне пришлось «простить» Oracle Source, настроив его на использование команды SQL в качестве ввода, а затем введя запрос с жестко закодированным сравнением дат. Как только я закончил создание потока данных, я вернулся к потоку управления, открыл свойства Data Flow Task и создал новую запись выражений. Я сопоставил переменную исходного запроса SSIS с [Oracle Source].[SqlCommand].
Также очень важно настроить BatchSize и размер буфера. По умолчанию источник Attunity Oracle использует BatchSize из 100. В отличие от этого, я обнаружил, что этот конкретный перенос хорошо работает с BatchSize из 120000
.
Суть сохраненного pro c, который объединяет данные, такова:
--This is an intermediate work table
TRUNCATE TABLE dbo.InvoiceWorkTable
--It gets populated from the SQL Server staging table where the raw Oracle data is stored
INSERT INTO dbo.InvoiceWorkTable
(
...
)
SELECT
...
FROM dbo.InvoiceTable_Staging
--Update records where the unique key matches and the HashVals don't match
UPDATE psd
SET
psd.* = iw.*
FROM dbo.ProductionSalesData psd
INNER JOIN dbo.InvoiceWorkTable iw
ON psd.DataSourceCode = iw.DataSourceCode
AND psd.InvoiceNumber = iw.InvoiceNum
AND psd.ItemNumber = iw.ItemNumber
AND psd.LineNumber = iw.LineNumber
AND psd.IsOpen = iw.IsOpen
WHERE iw.DataSourceCode = 'ABC'
AND iw.HashVal <> iw.HashVal
--Insert new records (i.e. those with unique keys not found in
INSERT INTO dbo.ProductionSalesData
(
...
)
SELECT
iw.*
FROM dbo.InvoiceWorkTable iw
LEFT JOIN dbo.ProductionSalesData psd
ON psd.DataSourceCode = iw.DataSourceCode
AND psd.InvoiceNumber = iw.InvoiceNum
AND psd.ItemNumber = iw.ItemNumber
AND psd.LineNumber = iw.LineNumber
AND psd.IsOpen = iw.IsOpen
WHERE iw.DataSourceCode = 'ECL'
AND psd.InvoiceNumber IS NULL
Я ничего не удаляю здесь, потому что это выполняется заданием очистки на конец месяца, которое выполняется отдельно. Я просто выполняю UPDATE для сопоставленных, но измененных записей, INSERT для новых, и все неизменное остается в одиночестве.
Отфильтровывая данные Oracle и используя хешированное кодирование записи для сравнения вместо сравнения каждого столбца, я избегаю очень требовательного к памяти варианта использования Lookup с сотнями миллионов значений. И при этом я не должен пытаться передавать эти идентификаторы между серверами (и платформами!).
Я буду sh Я мог бы сказать, что это был конец этого конкретного рабочего процесса, но это не так. Это очень ресурсоемкий процесс, и мы не можем позволить его запустить на нашем рабочем OLTP-сервере. Таким образом, мы запускаем все наши процессы ETL на совершенно другом сервере. Но нам также нужно, чтобы данные из Oracle были интегрированы в OLTP-сервер, как только они станут доступны. Но какой смысл в быстрой передаче сотен миллионов записей между Oracle и сервером ETL, если для переноса данных на другой сервер SQL все еще требуются часы? Выполнение «kill-and-fill» не сработает, потому что вставка занимает слишком много времени, и приложение слишком часто запрашивает данные. Я могу передавать данные между серверами, но получение новых данных в рабочую таблицу на OLTP-сервере не работает должным образом.
Так что после многих часов проб и ошибок я нашел учебное пособие о том, как используйте ALTER TABLE с SWITCH TO. Я не говорю, что это идеальное решение в каждом случае, так как есть соображения безопасности / разрешений, а также некоторые другие возможные предостережения. Если вам интересно, вам определенно следует прочитать больше из лучших источников, но в основном вы должны убедиться, что ваши исходные и конечные таблицы имеют одинаковое количество столбцов с одинаковыми именами, а также эквивалентные индексы. Наша таблица ProductionSalesData может иметь разные индексы между разными прогонами, поэтому мне пришлось создать этот pro c для динамического удаления и повторного создания индексов.
CREATE PROCEDURE [dbo].[proc_SwitchOutTables]
AS
DECLARE @CreateStatement NVARCHAR(MAX)
,@DropStatement NVARCHAR(MAX)
--this table will hold the current production data as a precaution
TRUNCATE TABLE dbo.ProductionSalesData_Old
--drop indexes on ProductionSalesData_Old
BEGIN
SELECT
DropStatement = 'DROP INDEX ' + QUOTENAME(i.name) + ' ON ' + QUOTENAME(t.name)
INTO #Drops
FROM sys.tables AS t
INNER JOIN sys.indexes AS i ON t.object_id = i.object_id
LEFT JOIN sys.dm_db_index_usage_stats AS u ON i.object_id = u.object_id AND i.index_id = u.index_id
WHERE t.is_ms_shipped = 0
AND i.type <> 0
AND t.name = 'ProductionSalesData_Old'
ORDER BY QUOTENAME(t.name), is_primary_key DESC
IF EXISTS (SELECT TOP 1 1 FROM #Drops)
BEGIN
DECLARE drop_cursor CURSOR FOR
SELECT DropStatement FROM #Drops
OPEN drop_cursor
FETCH NEXT FROM drop_cursor INTO @DropStatement
WHILE @@FETCH_STATUS = 0
BEGIN
EXEC(@DropStatement)
FETCH NEXT FROM drop_cursor INTO @DropStatement
END
--end loop
--clean up
CLOSE drop_cursor
DEALLOCATE drop_cursor
END
DROP TABLE #Drops
END
--recreate indexes on ProductionSalesData_Old based on indexes on ProductionSalesData
BEGIN
SELECT
CreateStatement =
'CREATE '
+ CASE WHEN i.type_desc = 'CLUSTERED' THEN 'CLUSTERED'
WHEN i.type_desc = 'NONCLUSTERED' AND is_unique=1 THEN 'UNIQUE NONCLUSTERED'
WHEN i.type_desc = 'NONCLUSTERED' AND is_unique=0 THEN 'NONCLUSTERED'
END
+ ' INDEX '
+ QUOTENAME(i.name)
+ ' ON '
+ QUOTENAME(t.name+'_Old')
+ ' ( '
+ STUFF(REPLACE(REPLACE((
SELECT QUOTENAME(c.name) + CASE WHEN ic.is_descending_key = 1 THEN ' DESC' ELSE '' END AS [data()]
FROM sys.index_columns AS ic
INNER JOIN sys.columns AS c ON ic.object_id = c.object_id AND ic.column_id = c.column_id
WHERE ic.object_id = i.object_id AND ic.index_id = i.index_id AND ic.is_included_column = 0
ORDER BY ic.key_ordinal
FOR XML PATH
), '<row>', ', '), '</row>', ''), 1, 2, '') + ' ) ' -- keycols
+ COALESCE(' INCLUDE ( ' +
STUFF(REPLACE(REPLACE((
SELECT QUOTENAME(c.name) AS [data()]
FROM sys.index_columns AS ic
INNER JOIN sys.columns AS c ON ic.object_id = c.object_id AND ic.column_id = c.column_id
WHERE ic.object_id = i.object_id AND ic.index_id = i.index_id AND ic.is_included_column = 1
ORDER BY ic.index_column_id
FOR XML PATH
), '<row>', ', '), '</row>', ''), 1, 2, '') + ' ) ', -- included cols
'')
+ COALESCE(' WHERE ' +
STUFF(REPLACE(REPLACE((
--SELECT QUOTENAME(c.name) AS [data()]
SELECT ic.filter_definition AS [data()]
FROM sys.indexes AS ic
WHERE ic.index_id = i.index_id
AND ic.object_id = i.object_id
AND ic.has_filter = 1
ORDER BY ic.index_id
FOR XML PATH
), '<row>', ', '), '</row>', ''), 1, 2, ''), -- filter
'')
INTO #Creates
FROM sys.tables AS t
INNER JOIN sys.indexes AS i ON t.object_id = i.object_id
LEFT JOIN sys.dm_db_index_usage_stats AS u ON i.object_id = u.object_id AND i.index_id = u.index_id
WHERE t.is_ms_shipped = 0
AND i.type <> 0
AND t.name = 'ProductionSalesData'
ORDER BY QUOTENAME(t.name)
,is_primary_key DESC
IF EXISTS (SELECT TOP 1 1 FROM #Creates)
BEGIN
DECLARE create_cursor CURSOR FOR
SELECT CreateStatement FROM #Creates
OPEN create_cursor
FETCH NEXT FROM create_cursor INTO @CreateStatement
WHILE @@FETCH_STATUS = 0
BEGIN
--PRINT @CreateStatement
EXEC(@CreateStatement)
FETCH NEXT FROM create_cursor INTO @CreateStatement
END
--end loop
--clean up
CLOSE create_cursor
DEALLOCATE create_cursor
END
DROP TABLE #Creates
END
--proc continues below
Последний бит про c ниже делает фактический SWITCH TO. Мы используем WITH ( WAIT_AT_LOW_PRIORITY(), чтобы дождаться, пока не поступит ожидающих запросов к нашей производственной таблице. Как только свободный участок очистится, текущая производственная таблица становится версией _Old, а версия _Staging, которую мы перенесли с сервера ETL, становится новой рабочей таблицей. SWITCH TO происходит в течение нескольких миллисекунд. Все, что делает сервер SQL, обновляет метаданные таблицы, в частности адрес памяти первого листа таблицы.
BEGIN TRAN
ALTER TABLE dbo.ProductionSalesData
SWITCH TO dbo.ProductionSalesData_Old
WITH ( WAIT_AT_LOW_PRIORITY ( MAX_DURATION = 1 MINUTES, ABORT_AFTER_WAIT = BLOCKERS ));
--Anyone who tries to query the table after the switch has happened and before
--the transaction commits will be blocked: we've got a schema mod lock on the table
ALTER TABLE dbo.ProductionSalesData_Staging
SWITCH TO dbo.ProductionSalesData;
COMMIT
GO