TensorFlow Data Validation (TFDV) peut analyser les données d'entraînement et de diffusion pour :
calculer des statistiques descriptives,
déduire un schéma ,
détecter les anomalies de données .
L'API principale prend en charge chaque élément de fonctionnalité, avec des méthodes pratiques qui s'appuient sur le dessus et peuvent être appelées dans le contexte des blocs-notes.
Calcul des statistiques descriptives des données
TFDV peut calculer des statistiques descriptives qui fournissent un aperçu rapide des données en termes d'entités présentes et de formes de leurs distributions de valeurs. Des outils tels que Facets Overview peuvent fournir une visualisation succincte de ces statistiques pour faciliter la navigation.
Par exemple, supposons que path pointe vers un fichier au format TFRecord (qui contient des enregistrements de type tensorflow.Example ). L'extrait de code suivant illustre le calcul des statistiques à l'aide de TFDV :
stats = tfdv.generate_statistics_from_tfrecord(data_location=path)
La valeur renvoyée est un tampon de protocole DatasetFeatureStatisticsList . L' exemple de notebook contient une visualisation des statistiques à l'aide de Facets Overview :
tfdv.visualize_statistics(stats)
L'exemple précédent suppose que les données sont stockées dans un fichier TFRecord . TFDV prend également en charge le format d'entrée CSV, avec une extensibilité pour d'autres formats courants. Vous pouvez trouver les décodeurs de données disponibles ici . De plus, TFDV fournit la fonction utilitaire tfdv.generate_statistics_from_dataframe pour les utilisateurs avec des données en mémoire représentées sous forme de pandas DataFrame.
En plus de calculer un ensemble de statistiques de données par défaut, TFDV peut également calculer des statistiques pour des domaines sémantiques (par exemple, images, texte). Pour activer le calcul des statistiques de domaine sémantique, transmettez un objet tfdv.StatsOptions avec enable_semantic_domain_stats défini sur True à tfdv.generate_statistics_from_tfrecord .
Exécution sur Google Cloud
En interne, TFDV utilise le cadre de traitement parallèle des données d' Apache Beam pour mettre à l'échelle le calcul des statistiques sur de grands ensembles de données. Pour les applications qui souhaitent s'intégrer plus profondément avec TFDV (par exemple, attacher la génération de statistiques à la fin d'un pipeline de génération de données, générer des statistiques pour les données dans un format personnalisé ), l'API expose également un Beam PTransform pour la génération de statistiques.
Pour exécuter TFDV sur Google Cloud, le fichier de roue TFDV doit être téléchargé et fourni aux nœuds de calcul Dataflow. Téléchargez le fichier de la roue dans le répertoire actuel comme suit :
pip download tensorflow_data_validation \
--no-deps \
--platform manylinux2010_x86_64 \
--only-binary=:all:
L'extrait de code suivant montre un exemple d'utilisation de TFDV sur Google Cloud :
import tensorflow_data_validation as tfdv
from apache_beam.options.pipeline_options import PipelineOptions, GoogleCloudOptions, StandardOptions, SetupOptions
PROJECT_ID = ''
JOB_NAME = ''
GCS_STAGING_LOCATION = ''
GCS_TMP_LOCATION = ''
GCS_DATA_LOCATION = ''
# GCS_STATS_OUTPUT_PATH is the file path to which to output the data statistics
# result.
GCS_STATS_OUTPUT_PATH = ''
PATH_TO_WHL_FILE = ''
# Create and set your PipelineOptions.
options = PipelineOptions()
# For Cloud execution, set the Cloud Platform project, job_name,
# staging location, temp_location and specify DataflowRunner.
google_cloud_options = options.view_as(GoogleCloudOptions)
google_cloud_options.project = PROJECT_ID
google_cloud_options.job_name = JOB_NAME
google_cloud_options.staging_location = GCS_STAGING_LOCATION
google_cloud_options.temp_location = GCS_TMP_LOCATION
options.view_as(StandardOptions).runner = 'DataflowRunner'
setup_options = options.view_as(SetupOptions)
# PATH_TO_WHL_FILE should point to the downloaded tfdv wheel file.
setup_options.extra_packages = [PATH_TO_WHL_FILE]
tfdv.generate_statistics_from_tfrecord(GCS_DATA_LOCATION,
output_path=GCS_STATS_OUTPUT_PATH,
pipeline_options=options)
Dans ce cas, le protocole de statistiques généré est stocké dans un fichier TFRecord écrit dans GCS_STATS_OUTPUT_PATH .
REMARQUE Lors de l'appel de l'une des fonctions tfdv.generate_statistics_... (par exemple, tfdv.generate_statistics_from_tfrecord ) sur Google Cloud, vous devez fournir un output_path . Spécifier Aucun peut provoquer une erreur.
Inférer un schéma sur les données
Le schéma décrit les propriétés attendues des données. Certaines de ces propriétés sont :
- quelles fonctionnalités sont censées être présentes
- leur genre
- le nombre de valeurs pour une caractéristique dans chaque exemple
- la présence de chaque fonctionnalité dans tous les exemples
- les domaines de fonctionnalités attendus.
En bref, le schéma décrit les attentes en matière de données "correctes" et peut donc être utilisé pour détecter des erreurs dans les données (décrites ci-dessous). De plus, le même schéma peut être utilisé pour configurer TensorFlow Transform pour les transformations de données. Notez que le schéma est censé être assez statique, par exemple, plusieurs ensembles de données peuvent se conformer au même schéma, alors que les statistiques (décrites ci-dessus) peuvent varier par ensemble de données.
Étant donné que l'écriture d'un schéma peut être une tâche fastidieuse, en particulier pour les ensembles de données comportant de nombreuses fonctionnalités, TFDV fournit une méthode pour générer une version initiale du schéma basée sur les statistiques descriptives :
schema = tfdv.infer_schema(stats)
En général, TFDV utilise une heuristique conservatrice pour déduire des propriétés de données stables à partir des statistiques afin d'éviter de suradapter le schéma à l'ensemble de données spécifique. Il est fortement conseillé de revoir le schéma déduit et de l'affiner au besoin , pour capturer toute connaissance du domaine sur les données que l'heuristique de TFDV aurait pu manquer.
Par défaut, tfdv.infer_schema déduit la forme de chaque entité requise, si value_count.min est égal à value_count.max pour l'entité. Définissez l'argument infer_feature_shape sur False pour désactiver l'inférence de forme.
Le schéma lui-même est stocké en tant que tampon de protocole de schéma et peut donc être mis à jour/modifié à l'aide de l'API standard de tampon de protocole. TFDV fournit également quelques méthodes utilitaires pour faciliter ces mises à jour. Par exemple, supposons que le schéma contienne la strophe suivante pour décrire une caractéristique de chaîne requise payment_type qui prend une seule valeur :
feature {
name: "payment_type"
value_count {
min: 1
max: 1
}
type: BYTES
domain: "payment_type"
presence {
min_fraction: 1.0
min_count: 1
}
}
Pour marquer que la fonctionnalité doit être renseignée dans au moins 50 % des exemples :
tfdv.get_feature(schema, 'payment_type').presence.min_fraction = 0.5
L' exemple de bloc-notes contient une visualisation simple du schéma sous forme de tableau, répertoriant chaque fonctionnalité et ses principales caractéristiques telles qu'encodées dans le schéma.
Vérification des données pour les erreurs
Étant donné un schéma, il est possible de vérifier si un jeu de données est conforme aux attentes définies dans le schéma ou s'il existe des anomalies de données . Vous pouvez rechercher des erreurs dans vos données (a) dans l'agrégat sur l'ensemble d'un ensemble de données en faisant correspondre les statistiques de l'ensemble de données avec le schéma, ou (b) en vérifiant les erreurs sur une base par exemple.
Faire correspondre les statistiques de l'ensemble de données à un schéma
Pour vérifier les erreurs dans l'agrégat, TFDV compare les statistiques de l'ensemble de données avec le schéma et marque les écarts. Par exemple:
# Assume that other_path points to another TFRecord file
other_stats = tfdv.generate_statistics_from_tfrecord(data_location=other_path)
anomalies = tfdv.validate_statistics(statistics=other_stats, schema=schema)
Le résultat est une instance du tampon de protocole Anomalies et décrit toutes les erreurs où les statistiques ne concordent pas avec le schéma. Par exemple, supposons que les données de other_path contiennent des exemples avec des valeurs pour la fonctionnalité payment_type en dehors du domaine spécifié dans le schéma.
Cela produit une anomalie
payment_type Unexpected string values Examples contain values missing from the schema: Prcard (<1%).
indiquant qu'une valeur hors domaine a été trouvée dans les statistiques dans < 1 % des valeurs de caractéristiques.
Si cela était prévu, le schéma peut être mis à jour comme suit :
tfdv.get_domain(schema, 'payment_type').value.append('Prcard')
Si l'anomalie indique vraiment une erreur de données, les données sous-jacentes doivent être corrigées avant de les utiliser pour la formation.
Les différents types d'anomalies pouvant être détectées par ce module sont listés ici .
L' exemple de cahier contient une visualisation simple des anomalies sous forme de tableau, répertoriant les fonctionnalités où des erreurs sont détectées et une brève description de chaque erreur.
Vérification des erreurs sur une base par exemple
TFDV offre également la possibilité de valider les données par exemple, au lieu de comparer les statistiques à l'échelle de l'ensemble de données avec le schéma. TFDV fournit des fonctions permettant de valider les données par exemple, puis de générer des statistiques récapitulatives pour les exemples anormaux trouvés. Par exemple:
options = tfdv.StatsOptions(schema=schema)
anomalous_example_stats = tfdv.validate_examples_in_tfrecord(
data_location=input, stats_options=options)
Le anomalous_example_stats qui validate_examples_in_tfrecord renvoie est un tampon de protocole DatasetFeatureStatisticsList dans lequel chaque ensemble de données se compose de l'ensemble d'exemples qui présentent une anomalie particulière. Vous pouvez l'utiliser pour déterminer le nombre d'exemples dans votre ensemble de données qui présentent une anomalie donnée et les caractéristiques de ces exemples.
Environnements de schéma
Par défaut, les validations supposent que tous les jeux de données d'un pipeline adhèrent à un seul schéma. Dans certains cas, il est nécessaire d'introduire de légères variations de schéma. Par exemple, les fonctionnalités utilisées comme libellés sont requises lors de la formation (et doivent être validées), mais sont manquantes lors de la diffusion.
Les environnements peuvent être utilisés pour exprimer de telles exigences. En particulier, les fonctionnalités du schéma peuvent être associées à un ensemble d'environnements à l'aide de default_environment, in_environment et not_in_environment.
Par exemple, si la fonctionnalité de conseils est utilisée comme étiquette dans la formation, mais manquante dans les données de diffusion. Sans environnement spécifié, il apparaîtra comme une anomalie.
serving_stats = tfdv.generate_statistics_from_tfrecord(data_location=serving_data_path)
serving_anomalies = tfdv.validate_statistics(serving_stats, schema)
Pour résoudre ce problème, nous devons définir l'environnement par défaut pour toutes les fonctionnalités sur "TRAINING" et "SERVING", et exclure la fonctionnalité "conseils" de l'environnement SERVING.
# All features are by default in both TRAINING and SERVING environments.
schema.default_environment.append('TRAINING')
schema.default_environment.append('SERVING')
# Specify that 'tips' feature is not in SERVING environment.
tfdv.get_feature(schema, 'tips').not_in_environment.append('SERVING')
serving_anomalies_with_env = tfdv.validate_statistics(
serving_stats, schema, environment='SERVING')
Vérification de l'asymétrie et de la dérive des données
En plus de vérifier si un jeu de données est conforme aux attentes définies dans le schéma, TFDV fournit également des fonctionnalités pour détecter :
- décalage entre les données d'entraînement et de diffusion
- dérive entre les différents jours de données d'entraînement
TFDV effectue cette vérification en comparant les statistiques de différents ensembles de données en fonction des comparateurs de dérive/inclinaison spécifiés dans le schéma. Par exemple, pour vérifier s'il existe une différence entre la fonctionnalité "payment_type" dans l'ensemble de données d'entraînement et de diffusion :
# Assume we have already generated the statistics of training dataset, and
# inferred a schema from it.
serving_stats = tfdv.generate_statistics_from_tfrecord(data_location=serving_data_path)
# Add a skew comparator to schema for 'payment_type' and set the threshold
# of L-infinity norm for triggering skew anomaly to be 0.01.
tfdv.get_feature(schema, 'payment_type').skew_comparator.infinity_norm.threshold = 0.01
skew_anomalies = tfdv.validate_statistics(
statistics=train_stats, schema=schema, serving_statistics=serving_stats)
REMARQUE La norme L-infini ne détectera l'inclinaison que pour les caractéristiques catégorielles. Au lieu de spécifier un seuil infinity_norm , spécifier un seuil jensen_shannon_divergence dans le skew_comparator détecterait l'asymétrie pour les caractéristiques numériques et catégorielles.
De même pour vérifier si un ensemble de données est conforme aux attentes définies dans le schéma, le résultat est également une instance du tampon de protocole Anomalies et décrit tout décalage entre les ensembles de données d'apprentissage et de service. Par exemple, supposons que les données de diffusion contiennent beaucoup plus d'exemples avec la fonctionnalité payement_type ayant la valeur Cash , cela produit une anomalie d'asymétrie
payment_type High L-infinity distance between serving and training The L-infinity distance between serving and training is 0.0435984 (up to six significant digits), above the threshold 0.01. The feature value with maximum difference is: Cash
Si l'anomalie indique vraiment un décalage entre les données d'entraînement et de diffusion, une enquête plus approfondie est nécessaire, car cela pourrait avoir un impact direct sur les performances du modèle.
L' exemple de bloc-notes contient un exemple simple de vérification des anomalies basées sur l'asymétrie.
La détection de la dérive entre différents jours de données d'entraînement peut être effectuée de la même manière
# Assume we have already generated the statistics of training dataset for
# day 2, and inferred a schema from it.
train_day1_stats = tfdv.generate_statistics_from_tfrecord(data_location=train_day1_data_path)
# Add a drift comparator to schema for 'payment_type' and set the threshold
# of L-infinity norm for triggering drift anomaly to be 0.01.
tfdv.get_feature(schema, 'payment_type').drift_comparator.infinity_norm.threshold = 0.01
drift_anomalies = tfdv.validate_statistics(
statistics=train_day2_stats, schema=schema, previous_statistics=train_day1_stats)
REMARQUE La norme L-infini ne détectera l'inclinaison que pour les caractéristiques catégorielles. Au lieu de spécifier un seuil infinity_norm , spécifier un seuil jensen_shannon_divergence dans le skew_comparator détecterait l'asymétrie pour les caractéristiques numériques et catégorielles.
Écriture d'un connecteur de données personnalisé
Pour calculer les statistiques de données, TFDV fournit plusieurs méthodes pratiques pour gérer les données d'entrée dans différents formats (par exemple TFRecord de tf.train.Example , CSV, etc.). Si votre format de données ne figure pas dans cette liste, vous devez écrire un connecteur de données personnalisé pour lire les données d'entrée et le connecter à l'API principale de TFDV pour calculer les statistiques de données.
L' API principale de TFDV pour le calcul des statistiques de données est un Beam PTransform qui prend une PCollection de lots d'exemples d'entrée (un lot d'exemples d'entrée est représenté sous la forme d'un Arrow RecordBatch) et génère une PCollection contenant un seul tampon de protocole DatasetFeatureStatisticsList .
Une fois que vous avez implémenté le connecteur de données personnalisé qui regroupe vos exemples d'entrée dans un Arrow RecordBatch, vous devez le connecter à l'API tfdv.GenerateStatistics pour calculer les statistiques de données. Prenez TFRecord de tf.train.Example par exemple. tfx_bsl fournit le connecteur de données TFExampleRecord , et ci-dessous est un exemple de la façon de le connecter avec l'API tfdv.GenerateStatistics .
import tensorflow_data_validation as tfdv
from tfx_bsl.public import tfxio
import apache_beam as beam
from tensorflow_metadata.proto.v0 import statistics_pb2
DATA_LOCATION = ''
OUTPUT_LOCATION = ''
with beam.Pipeline() as p:
_ = (
p
# 1. Read and decode the data with tfx_bsl.
| 'TFXIORead' >> (
tfxio.TFExampleRecord(
file_pattern=[DATA_LOCATION],
telemetry_descriptors=['my', 'tfdv']).BeamSource())
# 2. Invoke TFDV `GenerateStatistics` API to compute the data statistics.
| 'GenerateStatistics' >> tfdv.GenerateStatistics()
# 3. Materialize the generated data statistics.
| 'WriteStatsOutput' >> WriteStatisticsToTFRecord(OUTPUT_LOCATION))
Calcul de statistiques sur des tranches de données
TFDV peut être configuré pour calculer des statistiques sur des tranches de données. Le découpage peut être activé en fournissant des fonctions de découpage qui acceptent un Arrow RecordBatch et génèrent une séquence de tuples de forme (slice key, record batch) . TFDV fournit un moyen simple de générer des fonctions de découpage basées sur la valeur des caractéristiques qui peuvent être fournies dans le cadre de tfdv.StatsOptions lors du calcul des statistiques.
Lorsque le découpage est activé, le proto DatasetFeatureStatisticsList en sortie contient plusieurs protos DatasetFeatureStatistics , un pour chaque tranche. Chaque tranche est identifiée par un nom unique défini comme nom de l'ensemble de données dans le proto DatasetFeatureStatistics . Par défaut, TFDV calcule des statistiques pour l'ensemble de données global en plus des tranches configurées.
import tensorflow_data_validation as tfdv
from tensorflow_data_validation.utils import slicing_util
# Slice on country feature (i.e., every unique value of the feature).
slice_fn1 = slicing_util.get_feature_value_slicer(features={'country': None})
# Slice on the cross of country and state feature (i.e., every unique pair of
# values of the cross).
slice_fn2 = slicing_util.get_feature_value_slicer(
features={'country': None, 'state': None})
# Slice on specific values of a feature.
slice_fn3 = slicing_util.get_feature_value_slicer(
features={'age': [10, 50, 70]})
stats_options = tfdv.StatsOptions(
slice_functions=[slice_fn1, slice_fn2, slice_fn3])