Partager via


Tâche de flux de données

La tâche de flux de données encapsule le moteur de flux de données chargé de déplacer des données entre les sources et les destinations et permet à l'utilisateur de transformer, nettoyer et modifier les données au cours de leur déplacement. L'ajout d'une tâche de flux de données à un flux de contrôle de package permet au package d'extraire, de transformer et de charger des données.

Un flux de données est constitué d'au moins un composant de flux de données, mais il s'agit en général d'un ensemble de composants de flux de données connectés : des sources qui extraient des données, des transformations qui modifient, acheminent ou résument des données et des destinations qui chargent des données. Les composants sont connectés dans le flux de données par des chemins d'accès. Chaque chemin d'accès spécifie les deux composants qui en constituent le début et la fin. Pour plus d'informations, consultez Composants des flux de données.

Au moment de l'exécution, la tâche de flux de données crée un plan d'exécution à partir du flux de données et le moteur de flux de données exécute le plan. Vous pouvez créer une tâche de flux de données qui n'a aucun flux de données, mais la tâche s'exécute uniquement si elle inclut au moins un flux de données.

Le schéma suivant illustre une tâche de flux de données avec un flux de données.

Flux de données

Une tâche de flux de données peut contenir plusieurs flux de données. Si une tâche copie plusieurs datasets et si l'ordre dans lequel les données sont copiées n'est pas significatif, il peut être plus commode d'inclure plusieurs flux de données dans la tâche de flux de données. Par exemple, vous pourriez créer cinq flux de données, chacun copiant des données à partir d'un fichier plat vers une table de dimension différente dans un schéma d'entrepôt de données en étoile.

Toutefois, le moteur de flux de données détermine l'ordre d'exécution lorsqu'une tâche de flux de données contient plusieurs flux de données. Par conséquent, lorsque l'ordre est important, le package doit utiliser plusieurs tâches de flux de données, chacune contenant un flux de données. Vous pouvez ensuite appliquer des contraintes de précédence afin de contrôler l'ordre d'exécution des tâches.

Le schéma suivant illustre une tâche de flux de données contenant plusieurs flux de données.

Flux de données

Un package peut inclure plusieurs tâches de flux de données (c'est d'ailleurs souvent le cas des packages complexes). Par exemple, si un package requiert que des flux de données soient exécutés dans un ordre spécifique ou que d'autres tâches soient exécutées entre les flux de données, vous devez utiliser une tâche de flux de données distincte pour chaque flux de données.

La tâche de flux de données gère également les flux d'erreurs. Au moment de l'exécution, des erreurs au niveau des lignes peuvent se produire lorsque des composants de flux de données convertissent des données, effectuent des recherches ou évaluent des expressions. Par exemple, une colonne de données avec une valeur de chaîne ne peut pas être convertie en entier ou une expression tente de diviser par zéro. Ces deux opérations provoquent des erreurs et les lignes contenant ces erreurs peuvent être traitées séparément à l'aide d'un flux d'erreur. Pour plus d'informations sur l'utilisation de flux d'erreurs dans un flux de données de package, consultez Gestion des erreurs dans les données.

Pour insérer des données en bloc dans une base de données SQL Server à partir de fichiers texte, vous pouvez utiliser la tâche d'insertion en bloc plutôt qu'une tâche de flux de données et un flux de données. Cependant, la tâche d'insertion en bloc est incapable de transformer des données. Pour plus d'informations, consultez Tâche d'insertion en bloc.

Utilisation d'expressions de propriété avec des éléments de flux de données

Plusieurs composants de flux de données, notamment les sources, les transformations et les destinations, prennent en charge l'usage d'expressions de propriété dans certaines de leurs propriétés. Une expression de propriété est une expression qui remplace la valeur de la propriété lors du chargement du package. Au moment de l'exécution, le package utilise les valeurs de propriétés mises à jour. Les expressions sont construites à l'aide de la syntaxe d'expression Integration Services et peuvent inclure des fonctions, des opérateurs, des identificateurs et des variables Integration Services. Pour plus d'informations, consultez Informations de référence des expressions Integration Services, Utilisation d'expressions dans des packages et Utilisation des expressions de propriété dans des packages.

Si vous construisez un package dans Business Intelligence Development Studio, les propriétés de tous les composants de flux de données prenant en charge des expressions de propriété sont exposées dans la tâche de flux de données à laquelle elles appartiennent. Pour ajouter, modifier et supprimer les expressions de propriété des composants de flux de données, cliquez sur la tâche de flux de données, puis utilisez la fenêtre Propriétés ou l'éditeur de la tâche pour ajouter, modifier ou supprimer les expressions de propriété. Les expressions de propriété de la tâche de flux de données même sont gérées dans la fenêtre Propriétés.

Si le flux de données renferme des composants qui utilisent des expressions, celles-ci apparaissent également dans la fenêtre Propriétés. Pour visualiser les expressions, sélectionnez la tâche de flux de données à laquelle le composant appartient. Vous pouvez visualiser les propriétés par catégories ou par ordre alphabétique. Si vous optez pour l'affichage par catégories dans la fenêtre Propriétés, toutes les expressions non utilisées dans une propriété spécifique sont répertoriées dans la catégorie Divers. Si vous choisissez l'affichage par ordre alphabétique, les expressions s'affichent dans l'ordre du nom du composant de flux de données.

Entrées du journal

Integration Services propose un ensemble de journaux d'événements disponibles pour toutes les tâches. Integration Services fournit également des entrées de journal personnalisées pour de nombreuses tâches. Pour plus d'informations, consultez Implémentation de la journalisation dans les packages et Messages personnalisés pour la journalisation. La tâche de flux de données comprend les entrées de journal personnalisées suivantes :

Entrée du journal

Description

BufferSizeTuning

Indique que la tâche de flux de données a modifié la taille du tampon. L'entrée de journal décrit les raisons de cette modification de taille et indique la nouvelle taille temporaire du tampon.

OnPipelinePostEndOfRowset

Indique qu'un composant a reçu son signal de fin d'ensemble de lignes, défini par le dernier appel de la méthode ProcessInput. Une entrée est écrite pour chaque composant du flux de données qui traite l'entrée. L'entrée inclut le nom du composant.

OnPipelinePostPrimeOutput

Indique que le composant a terminé son dernier appel de la méthode PrimeOutput. Selon le flux de données, plusieurs entrées de journal peuvent être écrites. Si le composant est une source, cette entrée du journal signifie que le composant a terminé le traitement des lignes.

OnPipelinePreEndOfRowset

Indique qu'un composant est sur le point de recevoir son signal de fin d'ensemble de lignes, défini par le dernier appel de la méthode ProcessInput. Une entrée est écrite pour chaque composant du flux de données qui traite l'entrée. L'entrée inclut le nom du composant.

OnPipelinePrePrimeOutput

Indique que le composant est sur le point de recevoir son appel de la méthode PrimeOutput. Selon le flux de données, plusieurs entrées de journal peuvent être écrites.

OnPipelineRowsSent

Indique le nombre de lignes fournies à une entrée de composant par un appel de la méthode ProcessInput. L'entrée du journal inclut le nom du composant.

PipelineBufferLeak

Donne des informations sur tout composant qui maintient l'activité des tampons après la fermeture du gestionnaire de tampons. Si un tampon est toujours actif, cela signifie que des ressources des tampons n'ont pas été libérées et qu'elles peuvent provoquer des fuites de mémoire. L'entrée du journal fournit le nom du composant et l'ID du tampon.

PipelineComponentTime

Signale le temps (en millisecondes) que le composant a passé dans chacune de ses cinq étapes de traitement majeures : Validate, PreExecute, PostExecute, ProcessInput et ProcessOutput.

PipelineExecutionPlan

Indique le plan d'exécution du flux de données. Le plan d'exécution fournit des informations sur la manière dont les tampons sont envoyés aux composants. Ces informations, conjuguées à l'entrée PipelineExecutionTrees du journal, décrivent ce qui se passe dans la tâche de flux de données.

PipelineExecutionTrees

Indique les arborescences d'exécution de la disposition du flux de données. Le planificateur du moteur du flux de données utilise les arborescences pour construire le plan d'exécution du flux de données.

PipelineInitialization

Donne des informations d'initialisation relatives à la tâche. Ces informations incluent les répertoires à utiliser pour le stockage temporaire des données BLOB, la taille du tampon par défaut ainsi que le nombre de lignes contenues dans un tampon. Selon la configuration de la tâche de flux de données, plusieurs entrées de journal peuvent être écrites.

Ces entrées de journal fournissent de nombreuses informations sur l'exécution de la tâche de flux de données chaque fois que vous exécutez un package. D'une exécution à l'autre des packages, vous pouvez capturer des informations qui, au fil du temps, fournissent des éléments historiques importants sur le traitement effectué par la tâche, sur les problèmes pouvant affecter les performances et sur le volume de données géré par la tâche.

Pour plus d'informations sur la façon d'utiliser ces entrées de journal pour analyser et améliorer les performances du flux de données, consultez l'une des rubriques suivantes :

Exemples de messages d'une tâche de flux de données

Le tableau suivant répertorie des exemples de messages pour les entrées de journal d'un package très simple. Le package utilise une source OLE DB pour extraire les données d'une table, une transformation de tri pour trier les données et une destination OLE DB pour écrire les données dans une autre table.

Entrée du journal

Messages

BufferSizeTuning

Les lignes du type de mémoire tampon 0 risquent d'entraîner une taille de mémoire tampon supérieure à la configuration maximale. Les mémoires tampons de ce type ne comporteront que 9637 lignes.

Les lignes du type de mémoire tampon 2 risquent d'entraîner une taille de mémoire tampon supérieure à la configuration maximale. Les mémoires tampons de ce type ne comporteront que 9497 lignes.

Les lignes du type de mémoire tampon 3 risquent d'entraîner une taille de mémoire tampon supérieure à la configuration maximale. Les mémoires tampons de ce type ne comporteront que 9497 lignes.

OnPipelinePostEndOfRowset

Un composant recevra le signal de fin d'ensemble de lignes. : 1180 : Tri : 1181 : Entrée de tri

Un composant recevra le signal de fin d'ensemble de lignes. : 1291 : Destination OLE DB : 1304 : Entrée de destination OLE DB

OnPipelinePostPrimeOutput

Un composant a été retourné à partir de son appel PrimeOutput. : 1180 : Tri

Un composant a été retourné à partir de son appel PrimeOutput. : 1 : Source OLE DB

OnPipelinePreEndOfRowset

Un composant a terminé le traitement de toutes ses lignes. : 1180 : Tri : 1181 : Entrée de tri

Un composant a terminé le traitement de toutes ses lignes. : 1291 : Destination OLE DB : 1304 : Entrée de destination OLE DB

OnPipelinePrePrimeOutput

PrimeOutput sera appelé sur un composant. : 1180 : Tri

PrimeOutput sera appelé sur un composant. : 1 : Source OLE DB

OnPipelineRowsSent

Des lignes ont été fournies comme entrée à un composant de flux de données. : : 1185 : Sortie de source OLE DB : 1180 : Tri : 1181 : Entrée de tri : 76

Des lignes ont été fournies comme entrée à un composant de flux de données. : : 1308 : Sortie de tri : 1291 : Destination OLE DB : 1304 : Entrée de destination OLE DB : 76

PipelineComponentTime

Le composant "Calculate LineItemTotalCost" (3522) a passé 356 millisecondes dans ProcessInput.

Le composant "Sum Quantity and LineItemTotalCost" (3619) a passé 79 millisecondes dans ProcessInput.

Le composant "Calculate Average Cost" (3662) a passé 16 millisecondes dans ProcessInput.

Le composant "Sort by ProductID" (3717) a passé 125 millisecondes dans ProcessInput.

Le composant "Load Data" (3773) a passé 0 milliseconde dans ProcessInput.

Le composant"Extract Data" (3869) a passé 688 millisecondes dans PrimeOutput à remplir des tampons lors de la sortie "OLE DB Source Output" (3879).

Le composant "Sum Quantity and LineItemTotalCost" (3619) a passé 141 millisecondes dans PrimeOutput à remplir des tampons lors de la sortie "Aggregate Output 1" (3621).

Le composant "Sort by ProductID" (3717) a passé 16 millisecondes dans PrimeOutput à remplir des tampons lors de la sortie "Sort Output" (3719).

PipelineExecutionPlan

SourceThread0

Lecteurs : 1

Influences : 1180 1291

Liste des travaux de sortie

CreatePrimeBuffer de type 1 pour ID de sortie 11.

SetBufferListener : « WorkThread0 » pour ID d'entrée 1181

CreatePrimeBuffer de type 3 pour ID de sortie 12.

CallPrimeOutput sur OLE DB Source (1)

Fin de la liste des travaux de sortie

Fin SourceThread0

WorkThread0

Lecteurs : 1180

Influences : 1180 1291

Liste des travaux d'entrée, ID d'entrée 1181 (1 EORs attendus)

CallProcessInput sur ID d'entrée 1181 sur Sort (1180) pour type d'affichage 2

Fin de la liste des travaux d'entrée pour l'entrée 1181

Liste des travaux de sortie

CreatePrimeBuffer de type 4 pour ID de sortie 1182.

SetBufferListener : « WorkThread1 » pour ID d'entrée 1304

CallPrimeOutput sur Sort (1180)

Fin de la liste des travaux de sortie

Fin WorkThread0

WorkThread1

Lecteurs : 1291

Influences : 1291

Liste des travaux d'entrée, ID d'entrée 1304 (1 EORs attendus)

CallProcessInput sur ID d'entrée 1304 sur OLE DB Destination (1291) pour type d'affichage 5

Fin de la liste des travaux d'entrée pour l'entrée 1304

Liste des travaux de sortie

Fin de la liste des travaux de sortie

Fin WorkThread1

PipelineExecutionTrees

begin execution tree 0

output "Sortie de source OLE DB" (11)

input "Entrée de tri" (1181)

end execution tree 0

begin execution tree 1

output "Sortie d'erreur de source OLE DB" (12)

end execution tree 1

begin execution tree 2

output "Sortie de tri" (1182)

input "Entrée de destination OLE DB" (1304)

output "Sortie d'erreur de destination OLE DB" (1305)

end execution tree 2

PipelineInitialization

Aucun emplacement temporaire de stockage de données BLOB n'a été fourni. Le gestionnaire de mémoire tampon va prendre en compte les répertoires des variables d'environnement TEMP et TMP.

La taille de mémoire tampon par défaut est de 10485760 octets.

Chaque mémoire tampon comportera 10000 lignes par défaut.

Le flux de données ne supprimera pas les composants inutilisés, car sa propriété RunInOptimizedMode a la valeur False.

De nombreux événements de journal écrivent plusieurs entrées et les messages liés à de nombreuses entrées de journal contiennent des données complexes. Pour faciliter la compréhension et la communication du contenu des messages complexes, vous pouvez analyser leur texte. En fonction de l'emplacement des journaux, vous pouvez utiliser des instructions Transact-SQL ou un composant Script pour répartir le texte complexe sur plusieurs colonnes ou sous d'autres formats que vous jugez plus utiles.

Par exemple, le tableau suivant affiche le message « Des lignes ont été fournies comme entrée à un composant de flux de données. : : 1185 : Sortie de source OLE DB : 1180 : Tri : 1181 : Entrée de tri : 76 », analysé sur plusieurs colonnes. Le message a été écrit par l'événement OnPipelineRowsSent au moment où les lignes ont été transmises de la source OLE DB vers la transformation de tri.

Colonne

Description

Valeur

PathID

Valeur de la propriété ID du chemin d'accès entre la source OLE DB et la transformation de tri.

1185

PathName

Valeur de la propriété Name du chemin d'accès.

Sortie de source OLE DB

ComponentID

Valeur de la propriété ID de la transformation de tri.

1180

ComponentName

Valeur de la propriété Name de la transformation de tri.

Tri

InputID

Valeur de la propriété ID de l'entrée à la transformation de tri.

1181

InputName

Valeur de la propriété Name de l'entrée à la transformation de tri.

Entrée de tri

RowsSent

Nombre de lignes envoyées à l'entrée de la transformation de tri.

76

Sources

Les sources suivantes ont des propriétés pouvant être mises à jour par des expressions de propriété :

Pour plus d'informations, consultez Propriétés personnalisées des sources.

Transformations

Destinations

Les destinations suivantes ont des propriétés pouvant être mises à jour par des expressions de la propriété :

Pour plus d'informations, consultez Propriétés personnalisées des destinations.

Configuration de la tâche de flux de données

Vous pouvez définir des propriétés dans la fenêtre Propriétés ou par programme.

Pour plus d'informations sur la façon de définir ces propriétés dans la fenêtre Propriétés, cliquez sur la rubrique suivante :

Configuration de la tâche de flux de données par programme

Pour plus d'informations sur l'ajout par programme d'une tâche de flux de données à un package et sur la définition des propriétés de flux de données, cliquez sur la rubrique suivante :

Icône Integration Services (petite) Rester à jour avec Integration Services

Pour obtenir les derniers téléchargements, articles, exemples et vidéos de Microsoft, ainsi que des solutions sélectionnées par la communauté, visitez la page Integration Services sur MSDN ou TechNet :

Pour recevoir une notification automatique de ces mises à jour, abonnez-vous aux flux RSS disponibles sur la page.