The Business Intelligence Blog

Dans la suite des essais sur SMO, mon idée est de générer dynamiquement un package qui exporte les données en analysant les dépendances (fichier plat/ole db etc..). Sur la base de quelques conseils avisés, je me suis orienté vers un sample Microsoft qui détaille comment créer dynamiquement un package SSIS d'export.

Sur codeplex :
http://sqlsrvintegrationsrv.codeplex.com/
et plus précisement :
http://www.codeplex.com/SQLSrvIntegrationSrv/Release/ProjectReleases.aspx?ReleaseId=17647


J'ai trouvé le sample intéressant mais difficile à reprendre. Je vous invite à le consulter. De mon coté, je vais sur cette base, essayer d'illustrer la génération dynamique de package en détaillant les internes, en ayant comme objectif de produire le pipeline suivant :



Tout un programme...

Dans le précédent article, j'ai créé mon package et lui ajoutais un data flow (MainPipe).

Nous allons maintenant aborder l'aspect accès aux données dans ce billet. Au sein de l'ETL Microsoft, nous définissons des ConnectionManager dont l'objet est centraliser, de manière générique, les définitions et l'accès aux sources de données (et donc réduire la redondance de paramétrage)

(Tout le code source est disponible sur CodePlex : www.codeplex.com/BILAB, dans l'onglet source, répertoire POC/BILABHistory, solution BIHistory.sln. La classe concernée est MulticastExport.cs dans le sous projet BILAB.Technical.History.PackageBuilder)

Ajout des Gestionnaires de Connexion

Les gestionnaires de connexion (ConnectionManager) font l'objet d'une collection _p.Connections au sein de package. Vous pouvez en ajouter comme suit :

private runtime.ConnectionManager _sourceCM;

 

private void addSourceCM()

{

    _sourceCM = _p.Connections.Add("OLEDB");

    _sourceCM.Name = "AdventureDWSource";

    _sourceCM.ConnectionString = @"Provider=SQLOLEDB.1;Password=myuser;Persist Security Info=True;User ID=myuser;Initial Catalog=AdventureWorksDW;Data Source=localhost";

}

Les sources peuvent être FLATFILE, ADO, ODBC... fonction du besoin.

Ajout de la source au sein du DataFlow

Définir une connexion ne suffit pas en tant que tel à accéder aux données, il faut ajouter une source au sein de notre DataFlow, plus précisément au MainPipe :

private wrapper.IDTSComponentMetaData100 _oledbDatasource;

private wrapper.CManagedComponentWrapper _sourceInstance;

 

private void addSourceConnection()

   {

       _oledbDatasource = _mainpipe.ComponentMetaDataCollection.New();           

       _oledbDatasource.ComponentClassID = "DTSAdapter.OLEDBSource";

<…/>

Le pipeline SSIS est en fait composé d'objets de type ComponentMetadata qui représente chacune des tâches que vous vous pouvez glisser sur le designer SSIS.
Ce type vous permet d'accéder aux propriétés génériques (Commons) des tâches. Nous le "typons" en utilisant à nouveau les CLSID. 

Vous trouverez ici la référence des API Communes à toutes les tâches :

http://msdn.microsoft.com/en-us/library/ms136001.aspx

Plus précisément : http://msdn.microsoft.com/en-us/library/ms135950.aspx

 

Dans le modèle SSIS, nous devons aussi créer une instance de type « source OLE DB » (et non, ce n'est toujours pas fait...) avec la méthode _oleDbSource.Instanciate qui nous renvoie CManagedComponentWrapper qui comme son nom l'indique est un wrapper vers l'objet (COM encore) sous jacent :

<…/>

 

// Instanciate an OleDB source Instance for design time based on OleDB Source

_sourceInstance = _oledbDatasource.Instantiate();

// ProvideComponentProperties acts as a kind of constructor and build input, output ...//

_sourceInstance.ProvideComponentProperties();

 

<…/>

Si je refais un détour sur le développement de composants SSIS, je vous avais proposé il ya quelques temps une destination XML (qui ne passait pas par du DOM, mais par un XMLWriter) :

http://www.techheadbrothers.com/Articles.aspx/developpement-composant-integration-services-ssis

Et c'est en surchargeant la méthode ProvideComponentProperties que j'ajoutais les propriétés dynamiquement à mon composant.

http://www.techheadbrothers.com/Articles.aspx/developpement-composant-integration-services-ssis-page-3

Il faut donc bien comprendre que la méthode ProvideComponentProperties agit comme un véritable constructeur sur l'instance. Dès lors, nous aurons accès aux propriétés propres au PipelineComponent, et notamment lui la possibilité de lier notre source un connectionManager.

<…/>

// Relink runtime oledb component with the package connection manager defined before

            _oledbDatasource.RuntimeConnectionCollection[0].ConnectionManagerID = _sourceCM.ID;

            _oledbDatasource.RuntimeConnectionCollection[0].ConnectionManager

                            = runtime.DtsConvert.GetExtendedInterface(_sourceCM);

<…/>

Nous pouvons aussi spécifier le mode d'accès aux données ainsi que la requête (propriétés spécifiques):

<…/>

_sourceInstance.SetComponentProperty("AccessMode", 2);

_sourceInstance.SetComponentProperty("SqlCommand", "Select * from FactInternetSales");

   

_oledbDatasource.Name = "OLE select from Fact InternetSales";

_oledbDatasource.Description = "OLE select from Fact InternetSales";

<…/>

En complément au lien MSDN sur les propriétés communes, vous trouverez sur le lien suivant les propriétés propres à la source OLE DB :

http://msdn.microsoft.com/en-us/library/ms135923.aspx

Comme dernière étape à notre processus, nous allons faire en sorte que notre composant se connecte à sa source de données pour la valider et récupérer la liste des colonnes à extraire de la source comme suit:

<…/>

   _sourceInstance.AcquireConnections(null);

    _sourceInstance.ReinitializeMetaData();

    _sourceInstance.ReleaseConnections();

   

}

<…>

 

MulticastExport en action …

Enfin si je replace le tout dans l'appel :

public void Export()

        {

            _p = new runtime.Package();

            _p.Name = "MulticastExport";

            _p.CreatorComputerName = System.Environment.MachineName;

            _p.CreatorName = System.Environment.UserName;

   

            this.addDataflow();

            this.addSourceCM();

            this.addSourceConnection();

            //this.addMulticast();

            //this.addPdtSource();

            //this.addSort();

            //this.addMerge();

            //this.addFactDestination();

   

            runtime.Application a = new Microsoft.SqlServer.Dts.Runtime.Application();

            a.SaveToXml("ZBILAB.History.dtsx", _p, null);

        }

 

Nous retrouvons notre package avec une Connexion :

Et une source dans le dataflow :

Les propriétés « populées » avec nos paramètres :

Le vecteur colonne préparé suite à la validation de la source :

La suite …

A partir de cette source nous allons continuer à construire notre flux et ajouter un multicast qui nous permettra de faire un flux vers un fichier texte, et un flux vers une tâche Merge Join.

 

Nous avons donc rajouté notre source vers la table FactInternetSales, ainsi qu'un multicast après celle-ci. La boite multicast a pour objet de dupliquer les flux.

Dans un premier temps nous allons lui attacher un tri afin de faire une jointure avec les produits liés.

Et passer de :

Au dataflow suivant :

Ajout de la source de données Produit.

La première étape est logiquement le rajout de la source au produit. Quelques commentaires :

Tout comme la source OLE des faits, nous rajoutons dynamiquement un OLEDB source. Nous le lions au même ConnectionManager que la source de faits attendu qu'il s'agit de la même base de donnée (c'est 'utilité du CM).

La requête de cette source portera sur les produits :

Select * from DimProduct

Je ne reviens pas dans le détail car nous avons dé jà vu comment ajouter une source dans un dataflow.

 

private void addPdtSource()

{

    _pdtDatasource = _mainpipe.ComponentMetaDataCollection.New();

    //_oledbDatasource.ComponentClassID = "DTSAdapter.OleDbSource";

    _pdtDatasource.ComponentClassID = "DTSAdapter.OLEDBSource";

   

   

    // Instanciate an OleDB source Instance for design time based on OleDB Source

    _pdtSourceInstance = _pdtDatasource.Instantiate();

   

   

    // ProvideComponentProperties acts as a kind of constructor and build input, output ...

    _pdtSourceInstance.ProvideComponentProperties();

   

    // Relink runtime oledb component with the package connection manager defined before

    _pdtDatasource.RuntimeConnectionCollection[0].ConnectionManagerID = _sourceCM.ID;

    _pdtDatasource.RuntimeConnectionCollection[0].ConnectionManager

                    = runtime.DtsConvert.GetExtendedInterface(_sourceCM);

   

    /* Each dataflow component has properties :

     * ms-help://MS.VSCC.v90/MS.VSIPCC.v90/MS.SQLSVR.v10.en/s10is_5techref/html/065a337e-9a0d-4a5e-aa45-87da58fa37c3.htm

     * 1 - Common Properties :

     * ms-help://MS.VSCC.v90/MS.VSIPCC.v90/MS.SQLSVR.v10.en/s10is_5techref/html/51973502-5cc6-4125-9fce-e60fa1b7b796.htm

     * 2 - Custom Properties for oledb source :

     * ms-help://MS.VSCC.v90/MS.VSIPCC.v90/MS.SQLSVR.v10.en/s10is_5techref/html/1361a37f-d647-416b-bb96-542b5258f38d.htm#oledb

     */

   

    _pdtSourceInstance.SetComponentProperty("AccessMode", 2);

    _pdtSourceInstance.SetComponentProperty("SqlCommand", "Select * from DimProduct");

   

    _pdtDatasource.Name = "OLE select Product";

    _pdtDatasource.Description = "OLE select Product";

   

   

    _pdtSourceInstance.AcquireConnections(null);

    _pdtSourceInstance.ReinitializeMetaData();

    _pdtSourceInstance.ReleaseConnections();

   

}

 

Ajout des tris :

Petite nouveauté dans notre dataflow, les SORT Component. Nous en ajoutons 2 instances au sein de la méthode AddSort.

private void addSort()

        {

            // First sort

            _firstSort = _mainpipe.ComponentMetaDataCollection.New();

            _firstSort.ComponentClassID = "DTSTransform.Sort";

            wrapper.CManagedComponentWrapper firstSortInstance = _firstSort.Instantiate();

            firstSortInstance.ProvideComponentProperties();

            _firstSort.Name = "SRT First Sort";

   

Tout l'ajout d'une source OLEDB, l'ajout d'un tri se fait par l'ajout d'un ComponentMetadata (par la collection ad hoc).

Ce tri sera alors relatif à la table de fait via le multicast, une fois lié comme suit :

_mainpipe.PathCollection.New().AttachPathAndPropagateNotifications(

   _multicast1.OutputCollection[0], _firstSort.InputCollection[0]);

 

Le fait de rajouter ce lien crée dynamiquement un OutputCollection sur le multicast.

Il faut enfin spécifier sur quelle colonne nous trions notre flux :

wrapper.IDTSVirtualInput100 vInput =  _firstSort.InputCollection[0].GetVirtualInput();

 

wrapper.IDTSVirtualInputColumn100 vColumn = vInput.VirtualInputColumnCollection["ProductKey"];

 

wrapper.IDTSInputColumn100 col = firstSortInstance.SetUsageType(_firstSort.InputCollection[0].ID, vInput, vColumn.LineageID, wrapper.DTSUsageType.UT_READONLY);

 

   

// As sort component has custom properties, Column has also custom properties for sort

// see ms-help://MS.VSCC.v90/MS.VSIPCC.v90/MS.SQLSVR.v10.en/s10is_5techref/html/56f5df6a-56f6-43df-bca9-08476a3bd931.htm#sort

firstSortInstance.SetInputColumnProperty(_firstSort.InputCollection[0].ID, col.ID, "NewSortKeyPosition", 1);

 

Quelques commentaires sur ce qui précède :

Nous ne pouvons pas accéder directement aux listes de colonnes etc.… Nous le ferons via la Notion de VirtualInput qui donne une vue sur la collection. De même, nous retrouvons la clef « Productkey » par la collection VirtualInputcollection[object].

Nous faisons cela car nous souhaitons positionner un tri sur cette colonne. Tout comme nous avons vu que les composants avaient des propriétés communes, des propriétés spécifiques, telles que des propriétés sur les colonnes. Dans notre cas il s'agit de la propriété « NewSortKetPosition », qui va nous permettre d'indiquer que la colonne pointée par col.ID, sera la première clef de tri.

Pour finir, nous ajoutons dans le corps de cette méthode la création du deuxième sort en relation avec la source créées ci-dessus :

    // 2nd sort

    _secondSort = _mainpipe.ComponentMetaDataCollection.New();

    _secondSort.ComponentClassID = "DTSTransform.Sort";

    wrapper.CManagedComponentWrapper secondSortInstance = _secondSort.Instantiate();

    secondSortInstance.ProvideComponentProperties();

    _secondSort.Name = "SRT Second Sort";

   

    _mainpipe.PathCollection.New().AttachPathAndPropagateNotifications(

       _pdtDatasource.OutputCollection[0], _secondSort.InputCollection[0]);

   

    wrapper.IDTSVirtualInput100 vInput2 = _secondSort.InputCollection[0].GetVirtualInput();

    wrapper.IDTSVirtualInputColumn100 vColumn2 = vInput2.VirtualInputColumnCollection["ProductKey"];

    wrapper.IDTSInputColumn100 col2 = secondSortInstance.SetUsageType(_secondSort.InputCollection[0].ID, vInput2, vColumn2.LineageID, wrapper.DTSUsageType.UT_READONLY);

    // As sort component has custom properties, Column has also custom properties for sort

    // see ms-help://MS.VSCC.v90/MS.VSIPCC.v90/MS.SQLSVR.v10.en/s10is_5techref/html/56f5df6a-56f6-43df-bca9-08476a3bd931.htm#sort

    secondSortInstance.SetInputColumnProperty(_secondSort.InputCollection[0].ID, col2.ID, "NewSortKeyPosition", 1);

   

}

 

Le tour est complet. Reste à lier les deux flux sortant des tris sur un « Merge Join », dans le prochain post….