sql - tiempo - cómo insertar datos parellel en tres tablas diferentes

insertar datos en tablas relacionadas mysql java

Question

Tengo un procedimiento almacenado que insertará la mayor parte de los registros, ahora existe la posibilidad de insertar datos en 3 tablas en paralelo;

• Primera tabla que inserta 1 millón de registros.
• Segunda tabla que inserta 1,5 millones de registros.
• Tercera tabla que inserta 500k registros

Según mi conocimiento, la inserción del procedimiento está sucediendo una después de la otra.

Entonces, ¿cómo puedo implementar la carga en paralelo?

Answer 1

¿Son las tres tablas idénticas en estructura y contenido? De ser así, utilice la replicación transaccional / de fusión

Alternativamente, cree un disparador en la primera tabla para insertar en la segunda y tercera tabla

Answer 2

Suponiendo que desea tener el mismo valor de fecha de inserción para todas las inserciones, defina un parámetro de fecha establecido en la fecha actual como se muestra.

DECLARE @InsertDate as date
SET @InsertDate = GetDate()

Luego, pase el parámetro de fecha de inserción a su procedimiento insert insertado y actualice este procedimiento almacenado para usar esa entrada. Esto asegurará que se use el mismo valor de fecha de inserción para todas las inserciones.

EXEC dbo.InsertTables123 @p1 = @InsertDate

El parámetro de entrada @InsertDate también se puede asignar manualmente si se necesita algo distinto de la fecha actual.

Answer 3

Puede intentar crear tres trabajos y ejecutar scripts de inserción en paralelo de la siguiente manera:

DECLARE @jobId BINARY(16)
EXEC msdb.dbo.sp_add_job @job_name=N''Job1'',
@enabled=1,
@description=N''No description available.'',
@job_id = @jobId OUTPUT
EXEC msdb.dbo.sp_add_jobstep @job_id=@jobId, @step_name=N''Insert into First Table'',
@step_id=1,
@cmdexec_success_code=0,
@on_success_action=1,
@on_success_step_id=0,
@on_fail_action=2,
@on_fail_step_id=0,
@retry_attempts=0,
@retry_interval=0,
@os_run_priority=0, @subsystem=N''TSQL'',
@command=N''--Insert script for first table'',
@database_name=N''Test'',
@flags=0
EXEC msdb.dbo.sp_add_jobserver @job_id = @jobId, @server_name = N''(local)''
GO
DECLARE @jobId BINARY(16)
EXEC msdb.dbo.sp_add_job @job_name=N''Job2'',
@enabled=1,
@description=N''No description available.'',
@job_id = @jobId OUTPUT
EXEC msdb.dbo.sp_add_jobstep @job_id=@jobId, @step_name=N''Insert into second Table'',
@step_id=1,
@cmdexec_success_code=0,
@on_success_action=1,
@on_success_step_id=0,
@on_fail_action=2,
@on_fail_step_id=0,
@retry_attempts=0,
@retry_interval=0,
@os_run_priority=0, @subsystem=N''TSQL'',
@command=N''--Insert script for second table'',
@database_name=N''Test'',
@flags=0
EXEC msdb.dbo.sp_add_jobserver @job_id = @jobId, @server_name = N''(local)''
GO
DECLARE @jobId BINARY(16)
EXEC msdb.dbo.sp_add_job @job_name=N''Job3'',
@enabled=1,
@description=N''No description available.'',
@job_id = @jobId OUTPUT
EXEC msdb.dbo.sp_add_jobstep @job_id=@jobId, @step_name=N''Insert into Third Table'',
@step_id=1,
@cmdexec_success_code=0,
@on_success_action=1,
@on_success_step_id=0,
@on_fail_action=2,
@on_fail_step_id=0,
@retry_attempts=0,
@retry_interval=0,
@os_run_priority=0, @subsystem=N''TSQL'',
@command=N''--Insert script for third table'',
@database_name=N''Test'',
@flags=0
EXEC msdb.dbo.sp_add_jobserver @job_id = @jobId, @server_name = N''(local)''
GO
EXEC msdb.dbo.sp_start_job N''Job1'' ; --All will execute in parallel
EXEC msdb.dbo.sp_start_job N''Job2'' ;
EXEC msdb.dbo.sp_start_job N''Job3'' ;

Answer 4

Las declaraciones se ejecutan sincrónicamente dentro de un lote de T-SQL. Para ejecutar varias instrucciones de forma asincrónica y en paralelo desde un procedimiento almacenado, deberá usar múltiples conexiones de bases de datos concurrentes. Tenga en cuenta que la parte complicada con la ejecución asincrónica no solo determina cuándo se han completado todas las tareas, sino también si han tenido éxito o han fallado.

Método 1: paquete de SSIS

Cree un paquete SSIS para ejecutar las 3 sentencias SQL en paralelo. En SQL 2012 y versiones posteriores, ejecute el paquete utilizando los procedimientos almacenados del catálogo de SSIS. Pre-SQL 2012, deberá crear un trabajo de SQL Agent para el paquete y ejecutarlo con sp_start_job.

Deberá verificar el estado de ejecución de SSIS o el estado del trabajo del Agente SQL para determinar la finalización y el resultado de éxito / falla.

Método 2: Powershell y SQL Agent

Ejecute un trabajo del Agente SQL que ejecute un script de Powershell que ejecute las consultas en paralelo utilizando los trabajos en segundo plano de Powershell (comando Start-Job). La secuencia de comandos puede devolver un código de salida, cero para el éxito y distinto de cero para la falla, de modo que SQL Agent pueda determinar si tuvo éxito. Verifique el estado del trabajo del Agente SQL para determinar la finalización y el resultado correcto / incorrecto.

Método 3: múltiples trabajos del Agente SQL

Ejecute múltiples trabajos de SQL Agent al mismo tiempo, cada uno con un paso de trabajo T-SQL que contiene el script de importación. Compruebe el estado del trabajo del Agente SQL de cada trabajo para determinar la finalización y el resultado correcto / incorrecto.

Método 4: Service Broker Use un proc activado de cola para ejecutar los scripts de importación en paralelo. Esto puede ser obtuso si no ha utilizado el intermediario de servicios antes y es importante seguir los patrones examinados. He incluido un ejemplo para comenzar (reemplace THROW con RAISERROR para pre-SQL 2012). La base de datos debe tener habilitado Service Broker, que está habilitado por defecto pero apagado después de una restauración o adjuntar.

USE YourDatabase;
Go
--create proc that will be automatically executed (activated) when requests are waiting
CREATE PROC dbo.ExecuteTSqlTask
AS
SET NOCOUNT ON;
DECLARE
@TSqlJobConversationHandle uniqueidentifier = NEWID()
, @TSqlExecutionRequestMessage xml
, @TSqlExecutionResultMessage xml
, @TSqlExecutionResult varchar(10)
, @TSqlExecutionResultDetails nvarchar(MAX)
, @TSqlScript nvarchar(MAX)
, @TSqlTaskName sysname
, @RowsAffected int
, @message_type_name sysname;
WHILE 1 = 1
BEGIN
--get the next task to execute
WAITFOR (
RECEIVE TOP (1)
@TSqlJobConversationHandle = conversation_handle
, @TSqlExecutionRequestMessage = CAST(message_body AS xml)
, @message_type_name = message_type_name
FROM dbo.TSqlExecutionQueue
), TIMEOUT 1000;
IF @@ROWCOUNT = 0
BEGIN
--no work to do - exit
BREAK;
END;
IF @message_type_name = N''TSqlExecutionRequest''
BEGIN
--get task name and script
SELECT
@TSqlTaskName = @TSqlExecutionRequestMessage.value(''(/TSqlTaskName)[1]'', ''sysname'')
, @TSqlScript = @TSqlExecutionRequestMessage.value(''(/TSqlScript)[1]'', ''nvarchar(MAX)'');
--execute script
BEGIN TRY
EXEC sp_executesql @TSqlScript;
SET @RowsAffected = @@ROWCOUNT;
SET @TSqlExecutionResult = ''Completed'';
SET @TSqlExecutionResultDetails = CAST(@RowsAffected as varchar(10)) + '' rows affected'';
END TRY
BEGIN CATCH
SET @TSqlExecutionResult = ''Erred'';
SET @TSqlExecutionResultDetails =
''Msg '' + CAST(ERROR_NUMBER() AS varchar(10))
+ '', Level '' + CAST(ERROR_SEVERITY() AS varchar(2))
+ '', State '' + CAST(ERROR_STATE() AS varchar(10))
+ '', Line '' + CAST(ERROR_LINE() AS varchar(10))
+ '': '' + ERROR_MESSAGE();
END CATCH;
--send execution result back to initiator
SET @TSqlExecutionResultMessage = ''<TSqlTaskName /><TSqlExecutionResult /><TSqlExecutionResultDetails />'';
SET @TSqlExecutionResultMessage.modify(''insert text {sql:variable("@TSqlTaskName")} into (/TSqlTaskName)[1] '');
SET @TSqlExecutionResultMessage.modify(''insert text {sql:variable("@TSqlExecutionResult")} into (/TSqlExecutionResult)[1] '');
SET @TSqlExecutionResultMessage.modify(''insert text {sql:variable("@TSqlExecutionResultDetails")} into (/TSqlExecutionResultDetails)[1] '');
SEND ON CONVERSATION @TSqlJobConversationHandle
MESSAGE TYPE TSqlExecutionResult
(@TSqlExecutionResultMessage);
END
ELSE
BEGIN
IF @message_type_name = N''TSqlJobComplete''
BEGIN
--service has ended conversation so we''re not going to get any more execution requests
END CONVERSATION @TSqlJobConversationHandle;
END
ELSE
BEGIN
END CONVERSATION @TSqlJobConversationHandle WITH ERROR = 1 DESCRIPTION = ''Unexpected message type received by ExecuteTSqlTask'';
RAISERROR(''Unexpected message type received (%s) by ExecuteTSqlTask'', 16, 1, @message_type_name);
END;
END;
END;
GO
CREATE QUEUE dbo.TSqlResultQueue;
CREATE QUEUE dbo.TSqlExecutionQueue
WITH STATUS=ON,
ACTIVATION (
STATUS = ON
, PROCEDURE_NAME = dbo.ExecuteTSqlTask
, MAX_QUEUE_READERS = 3 --max number of concurrent activated proc instances
, EXECUTE AS OWNER
);
CREATE MESSAGE TYPE TSqlExecutionRequest VALIDATION = WELL_FORMED_XML;
CREATE MESSAGE TYPE TSqlExecutionResult VALIDATION = WELL_FORMED_XML;
CREATE MESSAGE TYPE TSqlJobComplete VALIDATION = WELL_FORMED_XML;
CREATE CONTRACT TSqlExecutionContract (
TSqlExecutionRequest SENT BY INITIATOR
, TSqlJobComplete SENT BY INITIATOR
, TSqlExecutionResult SENT BY TARGET
);
CREATE SERVICE TSqlJobService ON QUEUE dbo.TSqlResultQueue ([TSqlExecutionContract]);
CREATE SERVICE TSqlExecutorService ON QUEUE dbo.TSqlExecutionQueue ([TSqlExecutionContract]);
GO
CREATE PROC dbo.ExecuteParallelImportScripts
AS
SET NOCOUNT ON;
DECLARE
@TSqlJobConversationHandle uniqueidentifier
, @TSqlExecutionRequestMessage xml
, @TSqlExecutionResultMessage xml
, @TSqlExecutionResult varchar(10)
, @TSqlExecutionResultDetails nvarchar(MAX)
, @TSqlTaskName sysname
, @CompletedCount int = 0
, @ErredCount int = 0
, @message_type_name sysname;
DECLARE @TsqlTask TABLE(
TSqlTaskName sysname NOT NULL PRIMARY KEY
, TSqlScript nvarchar(MAX) NOT NULL
);
BEGIN TRY
--insert a row for each import task
INSERT INTO @TsqlTask(TSqlTaskName, TSqlScript)
VALUES(N''ImportScript1'', N''INSERT INTO dbo.Table1 SELECT * FROM dbo.Table1Staging;'');
INSERT INTO @TsqlTask(TSqlTaskName, TSqlScript)
VALUES(N''ImportScript2'', N''INSERT INTO dbo.Table2 SELECT * FROM dbo.Table2Staging;'');
INSERT INTO @TsqlTask(TSqlTaskName, TSqlScript)
VALUES(N''ImportScript3'', N''INSERT INTO dbo.Table3 SELECT * FROM dbo.Table3Staging;'');
--start a conversation for this import process
BEGIN DIALOG CONVERSATION @TsqlJobConversationHandle
FROM SERVICE TSqlJobService
TO SERVICE ''TSqlExecutorService'', ''CURRENT DATABASE''
ON CONTRACT TSqlExecutionContract
WITH ENCRYPTION = OFF;
--send import tasks to executor service for parallel execution
DECLARE JobTasks CURSOR LOCAL FAST_FORWARD FOR
SELECT (SELECT TSqlTaskName, TSqlScript
FROM @TsqlTask AS task
WHERE task.TSqlTaskName = job.TSqlTaskName
FOR XML PATH(''''), TYPE) AS TSqlExecutionRequest
FROM @TsqlTask AS job;
OPEN JobTasks;
WHILE 1 = 1
BEGIN
FETCH NEXT FROM JobTasks INTO @TSqlExecutionRequestMessage;
IF @@FETCH_STATUS = -1 BREAK;
SEND ON CONVERSATION @TSqlJobConversationHandle
MESSAGE TYPE TSqlExecutionRequest
(@TSqlExecutionRequestMessage);
END;
CLOSE JobTasks;
DEALLOCATE JobTasks;
--get each parallel task execution result until all are complete
WHILE 1 = 1
BEGIN
--get next task result
WAITFOR (
RECEIVE TOP (1)
@TSqlExecutionResultMessage = CAST(message_body AS xml)
, @message_type_name = message_type_name
FROM dbo.TSqlResultQueue
WHERE conversation_handle = @TSqlJobConversationHandle
), TIMEOUT 1000;
IF @@ROWCOUNT <> 0
BEGIN
IF @message_type_name = N''TSqlExecutionResult''
BEGIN
--get result of import script execution
SELECT
@TSqlTaskName = @TSqlExecutionResultMessage.value(''(/TSqlTaskName)[1]'', ''sysname'')
, @TSqlExecutionResult = @TSqlExecutionResultMessage.value(''(/TSqlExecutionResult)[1]'', ''varchar(10)'')
, @TSqlExecutionResultDetails = COALESCE(@TSqlExecutionResultMessage.value(''(/TSqlExecutionResultDetails)[1]'', ''nvarchar(MAX)''), N'''');
RAISERROR(''Import task %s %s: %s'', 0, 0, @TSqlTaskName, @TSqlExecutionResult, @TSqlExecutionResultDetails) WITH NOWAIT;
IF @TSqlExecutionResult = ''Completed''
BEGIN
SET @CompletedCount += 1;
END
ELSE
BEGIN
SET @ErredCount += 1;
END;
--remove task from tracking table after completion
DELETE FROM @TSqlTask
WHERE TSqlTaskName = @TSqlTaskName;
IF NOT EXISTS(SELECT 1 FROM @TsqlTask)
BEGIN
--all tasks are done - send TSqlJobComplete message to instruct executor service to end conversation
SEND ON CONVERSATION @TSqlJobConversationHandle
MESSAGE TYPE TSqlJobComplete;
END
END
ELSE
BEGIN
IF @message_type_name = N''http://schemas.microsoft.com/SQL/ServiceBroker/EndDialog''
BEGIN
--executor service has ended conversation so we''re done
END CONVERSATION @TSqlJobConversationHandle;
BREAK;
END
ELSE
BEGIN
END CONVERSATION @TSqlJobConversationHandle WITH ERROR = 1 DESCRIPTION = ''Unexpected message type received by ExecuteParallelInserts'';
RAISERROR(''Unexpected message type received (%s) by ExecuteParallelInserts'', 16, 1, @message_type_name);
END;
END
END;
END;
RAISERROR(''Import processing completed. CompletedCount=%d, ErredCount=%d.'', 0, 0, @CompletedCount, @ErredCount);
END TRY
BEGIN CATCH
THROW;
END CATCH;
GO
--execute import scripts in parallel
EXEC dbo.ExecuteParallelImportScripts;
GO

Answer 5

Para su procedimiento, supongo que tiene TableName y la ubicación del archivo como parámetros.

Si tiene un archivo grande que tiene 3 millones de registros, debe dividir el archivo en 3 pequeños (si conoce otro idioma que no sea sql), después de eso, puede abrir 3 consolas de servidores Sql y llamar el procedimiento en cada consola. Hará que la inserción sea paralela. O conoce otros lenguajes de programación, puede usar varios hilos para llamar al procedimiento.