Blog de Eduardo Sojo

Se pueden crear paquetes de SQL Server Integration Services (SSIS) programáticamente con .NET, haciendo uso de las librerías que incluye SQL Server 2008. En este post se explica cómo crear data flow task y sus elementos internos.

Serie Creando paquetes de SSIS con .NET

• Agregando elementos básicos
• Creando data flow Task y elementos internos
• Creando SQL Task y File System Task
• Creando Execute Package Task y Execute Process Task

Una vez que agregamos los elementos básicos como creación del paquete y los objetos de conexión, podemos seguir agregando nuevos elementos al paquete que estamos creando usando C# y .NET (puedes guiarte con el primer post de esta serie - Agregando elementos básicos)

Tomaremos como referencia los elementos creados en el primer post de esta serie, para seguir agregándole objetos al paquete que se quiere crear. Para este post crearemos un data flow y algunos elementos internos, tomando como ejemplo un escenario en donde tomaremos data de un archivo de texto y la pasaremos a una tabla en SQL Server.

Creando Dataflow Task

El dataflow task es el objeto que contendrá todo los flujos de trabajo para trabajar con los datos. Para crear el objeto Dataflow Task utilizaremos el siguiente código:

Executable DataFlowExe = pkg.Executables.Add("STOCK:PipelineTask");
TaskHost thMainPipe = DataFlowExe as TaskHost;
thMainPipe.Name = "Carga de tabla";
MainPipe dfTask = thMainPipe.InnerObject as MainPipe;
PrecedenceConstraint constraint = pkg.PrecedenceConstraints.Add(SQLTaskExe, DataFlowExe);

Ya teniendo creado el dataflow task, podemos agregar los objetos internos que conforman el flujo de trabajo a construir.

Creando Flat File Source

Este objeto es que nos permite conectarnos con el origen de los datos para extraer los registros, este origen de datos debe ser un objeto de conexión de tipo Flat File en el cual se define el comportamiento de la información dentro del archivo.

Para crear este tipo de objetos usamos el siguiente código:

IDTSComponentMetaData100 ffSource = dfTask.ComponentMetaDataCollection.New();
ffSource.ComponentClassID = "DTSAdapter.FlatFileSource.2";
ffSource.Name = "FlatFileSource";
ffSource.Description = "Flat file source";
CManagedComponentWrapper ffSrcComponent = ffSource.Instantiate();
ffSrcComponent.ProvideComponentProperties();
ffSrcComponent.SetComponentProperty("RetainNulls", true);

//Asociamos al objeto de conexión
if (ffSource.RuntimeConnectionCollection.Count > 0)
{
ffSource.RuntimeConnectionCollection[0].ConnectionManager = DtsConvert.GetExtendedInterface(pkg.Connections["TXT.Archivo"]);
ffSource.RuntimeConnectionCollection[0].ConnectionManagerID = pkg.Connections["TXT.Archivo"].ID;
}
ffSrcComponent.AcquireConnections(null);
ffSrcComponent.ReinitializeMetaData();
ffSrcComponent.ReleaseConnections();

//Opcional, podemos definir el comportamiento de los campos al momento de ocurrir un error o un truncamiento de los datos
for (int i = 0; i < ffSource.OutputCollection[0].OutputColumnCollection.Count; i++)
{
ffSource.OutputCollection[0].OutputColumnCollection[i].ErrorRowDisposition = DTSRowDisposition.RD_RedirectRow;
ffSource.OutputCollection[0].OutputColumnCollection[i].TruncationRowDisposition = DTSRowDisposition.RD_RedirectRow;
}

Creando SQL Destination

El siguiente paso es crear el destino de los datos, para este caso es una tabla de SQL Server. Para crear este objeto y definir su comportamiento usamos el siguiente código:

IDTSComponentMetaData100 SQLDestination = dfTask.ComponentMetaDataCollection.New();
SQLDestination.ComponentClassID = "DTSAdapter.OLEDBDestination.2";
SQLDestination.Name = "OLEDBDestination";
SQLDestination.Description = "SQL destination";
CManagedComponentWrapper SQLDestComponent = SQLDestination.Instantiate();
SQLDestComponent.ProvideComponentProperties();

//Asociamos con el objeto de conexión de SQL Server
if (SQLDestination.RuntimeConnectionCollection.Count > 0)
{
SQLDestination.RuntimeConnectionCollection[0].ConnectionManager = DtsConvert.GetExtendedInterface(pkg.Connections["OLEDB"]);
SQLDestination.RuntimeConnectionCollection[0].ConnectionManagerID = pkg.Connections["OLEDB"].ID;
}

//Configuramos algunas propiedades del objeto
SQLDestComponent.SetComponentProperty("AccessMode", 3); //Equivale a: Table or view - fast load
SQLDestComponent.SetComponentProperty("OpenRowset", "NombreTabla");
SQLDestComponent.SetComponentProperty("FastLoadOptions", "TABLOCK");
SQLDestComponent.AcquireConnections(null);

//Reinicializamos la metadata del objeto
SQLDestComponent.ReinitializeMetaData();

Conexión de ambos objetos

Finalmente debemos unir ambos objetos, o mejor dicho, trazar por código la conexión de los objetos mapeando la salida de uno con la entrada del otro.

Para realizar esta actividad aplicamos el siguiente código:

//Agregamos el conector
IDTSPath100 path = dfTask.PathCollection.New();
path.AttachPathAndPropagateNotifications(ffSource.OutputCollection[0], SQLDestination.InputCollection[0]);
// Obtenemos el input por defecto del objeto.
IDTSInput100 input = SQLDestination.InputCollection[0];
IDTSVirtualInput100 vInput = input.GetVirtualInput();

foreach (IDTSVirtualInputColumn100 vColumn in vInput.VirtualInputColumnCollection)
{
// Llamamos al método SetUsageType del objeto destino para agregar cada virtual input column como un input column.
SQLDestComponent.SetUsageType(input.ID, vInput, vColumn.LineageID, DTSUsageType.UT_READWRITE);
}
foreach (IDTSInputColumn100 inColumn in SQLDestination.InputCollection[0].InputColumnCollection)
{
// creamos el mapeo
IDTSExternalMetadataColumn100 exColumn = SQLDestination.InputCollection[0].ExternalMetadataColumnCollection[inColumn.Name];
exColumn.Name = inColumn.Name;
SQLDestComponent.MapInputColumn(SQLDestination.InputCollection[0].ID, inColumn.ID, exColumn.ID);
}
SQLDestComponent.ReleaseConnections();

De esta forma completamos el flujo de trabajo que procesaría los registros contenidos en un archivo de texto para enviarlos a una tabla de SQL Server. En la siguiente entrega estaré mostrando un poco cómo crear SQL Task y File System Task.

Nos leemos,