<jupyter_start><jupyter_text>Tensorflow Validation with Tensorflow PipelineWe have seen how to generate baseline statistics and schema, and now we need to include the data validation as part of our ML workflow. Tensorflow Pipeline provides components to let you easily do that, and this is what we are going to do now. Specifically, we will build a data pipeline by:- Extracting data from BigQuery using Tensorflow Pipeline.- Generating the statistics of the dataset.- Creating a schema as per the domain knowledge.- Apply schema to a new dataset.- Check for anomalies. DatasetThis notebook uses the same dataset as the previous lab: Chicago crime data.[the previous lab](https://www.freeldom.com/article/TFDV%20-Lab-1-1622385519244)This notebook uses [Chicago crime data](https://data.cityofchicago.org/) published as a public dataset in BigQuery. This dataset reflects reported incidents of crime that occurred in the City of Chicago from 2001 to the present, minus the most recent seven days. The data will be extracted with the following columns:- **date**: Date when the incident occurred.- **iucr**: The Illinois Unifrom Crime Reporting code.- **primary_type**: The primary description of the IUCR code.- **location_description**: Description of the location where the incident occurred.- **arrest**: Indicates whether an arrest was made.- **domestic**: Indicates whether the incident was domestic-related as defined by the Illinois Domestic Violence Act.- **district**: Indicates the police district where the incident occurred. - **ward**: The ward (City Council district) where the incident occurred.- **fbi_code**: Indicates the crime classification.- **year**: Year the incident occurred. Installing dependencies<jupyter_code>!pip install tfx<jupyter_output><empty_output><jupyter_text>Imports<jupyter_code>import tensorflow as tf
import tfx

import tensorflow_data_validation as tfdv
import tensorflow_transform as tft
from tfx.orchestration.experimental.interactive.interactive_context import InteractiveContext
from tfx.extensions.google_cloud_big_query.example_gen.component import BigQueryExampleGen
from tensorflow_metadata.proto.v0 import schema_pb2

from tfx.components import ExampleValidator
from tfx.components import SchemaGen
from tfx.components import StatisticsGen
from tfx.components import ImporterNode

from tfx.types import standard_artifacts

from datetime import datetime

CHICAGO_CRIME_TABLE = 'bigquery-public-data.chicago_crime.crime'

GCS_BUCKET = "" # Set your GCS bucket
PROJECT_ID = '' # Set your GCP Project Id
REGION = '' # Set the region for Dataflow jobs

#PIPELINE_ID = datetime.now().strftime('%Y%m%d-%H%M%S')<jupyter_output><empty_output><jupyter_text>Setup Interactive ContextWe will set up the Interactive Context so we can manually execute the pipeline components from the notebook.<jupyter_code>PIPELINE_ROOT = './pipeline'

# Declare the InteractiveContext and use a local sqlite file as the metadata store.
context = InteractiveContext(pipeline_root=PIPELINE_ROOT)<jupyter_output>WARNING:absl:InteractiveContext metadata_connection_config not provided: using SQLite ML Metadata database at ./pipeline/metadata.sqlite.<jupyter_text>Extracting data from BigQueryThe first step of the pipeline is to extract the data from BigQuery. In the previous lab, we did it using our own DataFlow pipeline. This time we will use the tfx component [BigQueryExampleGen](https://www.tensorflow.org/tfx/api_docs/python/tfx/extensions/google_cloud_big_query/example_gen/component/BigQueryExampleGen), but the query remains the same.<jupyter_code>def generate_query(year_from= None, year_to= None, limit= None) -> str:
    query = f"""
        SELECT 
            FORMAT_DATE('%Y',  CAST(date AS DATE)) AS crime_year,
            FORMAT_DATE('%b',  CAST(date AS DATE)) AS crime_month,
            FORMAT_DATE('%d',  CAST(date AS DATE)) AS crime_day, 
            FORMAT_DATE('%a',  CAST(date AS DATE)) AS crime_day_of_week, 
            iucr,
            primary_type,
            location_description,
            CAST(domestic AS INT64) AS domestic,
            district,
            ward,
            fbi_code,
            CAST(arrest AS INT64) AS arrest,
        FROM 
          {CHICAGO_CRIME_TABLE}
        """
    if year_from:
        query += f"WHERE year >= {year_from}"
        if year_to:
            query += f" AND year <= {year_to} \n"
    if limit:
        query  += f"LIMIT {limit}"
        
    return query
# Some parameters needed for BigQuery, but remember we run the pipeline locally, so we will extract only 2019 data
example_config = [
    '--project', PROJECT_ID,
    '--temp_location', f"gs://{GCS_BUCKET}/tfdv/pipeline/{PIPELINE_ID}/",
    '--region', REGION
]

example_gen = BigQueryExampleGen(query=generate_query(2019, 2019))
context.run(example_gen, beam_pipeline_args=example_config)<jupyter_output><empty_output><jupyter_text>Computing statisticsNext, let's compute the statistics to observe and analyze characteristics of our data. We will use the component [StatisticsGen](https://www.tensorflow.org/tfx/guide/statsgen) to do that. We need to feed in the data we have extracted previously. We use the component outputs that has the path of where the data has been extracted.<jupyter_code>statistics_gen = StatisticsGen(
      examples=example_gen.outputs['examples']
)
context.run(statistics_gen)
# Let's show the statistics
context.show(statistics_gen.outputs['statistics'])<jupyter_output><empty_output><jupyter_text>Inferring the schemaAs usual, we need to create a schema that we will use to validate incoming datasets during training and serving. The TFX component to do that is named [SchemaGen](https://www.tensorflow.org/tfx/guide/schemagen) component.<jupyter_code>schema_gen = SchemaGen(
    statistics=statistics_gen.outputs['statistics']
)
context.run(schema_gen)
context.show(schema_gen.outputs['schema'])<jupyter_output><empty_output><jupyter_text>We also know that we must try to enhance the inferred schema with the domain knowledge that we have. Let's apply the same constraints as the ones of the first lab. First, we get the path where our schema has been saved, and load it.<jupyter_code>schema_uri = schema_gen.outputs['schema']._artifacts[0].uri
baseline_schema = tfdv.load_schema_text(schema_uri + '/schema.pbtxt')

# Setting domain
tfdv.set_domain(baseline_schema, "domestic", schema_pb2.IntDomain(min=0, max=1))
tfdv.set_domain(baseline_schema, "district", schema_pb2.IntDomain(min=1, max=31))

# Adding crime type
primary_types = tfdv.get_domain(baseline_schema, 'primary_type')
primary_types.value.append('RITUALISM')

# Setting up environments
baseline_schema.default_environment.append('TRAINING')
baseline_schema.default_environment.append('SERVING')
tfdv.get_feature(baseline_schema, 'arrest').not_in_environment.append('SERVING')
tfdv.display_schema(schema=baseline_schema)
# Let's save the updated schema
schema_folder = f'{PIPELINE_ROOT}/enhanced_schema'
!mkdir -p {schema_folder}
schema_filename = f'{schema_folder}/schema.pbtxt'

# Save the curated schema to the said file
tfdv.write_schema_text(baseline_schema, schema_filename)<jupyter_output><empty_output><jupyter_text>Building the pipelineWe have our baseline schema and statistics, and the goal is to use them for our next training job or our next serving job. In the next cells, we are going to use TFX components to import our baseline statistics, extract new dataset, compute the statistics and compare with our baseline. In a real-world situation, if we identify anomalies, the ML pipeline should raise an alert. Otherwise, we would continue with preprocessing, training and serving.   Import new schemaWe use [ImporterNode](https://www.tensorflow.org/tfx/guide/statsgenusing_the_statsgen_component_with_a_schema) to import the schema.<jupyter_code>imported_baseline_schema = ImporterNode(
    source_uri=schema_folder,
    artifact_type=standard_artifacts.Schema
).with_id('baseline_schema')
context.run(imported_baseline_schema, enable_cache=False)<jupyter_output><empty_output><jupyter_text>Extract data from 2020Let's extract 2020 data and compute the statistics. We will then check for anomalies using our baseline schema and [ExampleValidator](https://www.tensorflow.org/tfx/guide/exampleval) component.<jupyter_code># Some parameters needed for BigQuery, but remember we run the pipeline locally, so we will extract only 2019 data
example_config = [
    '--project', PROJECT_ID,
    '--temp_location', f"gs://{GCS_BUCKET}/tfdv/pipeline/{PIPELINE_ID}/",
    '--region', REGION
]

example_gen = BigQueryExampleGen(query=generate_query(2020, 2020))
context.run(example_gen, beam_pipeline_args=example_config)

statistics_gen = StatisticsGen(
      examples=example_gen.outputs['examples']
)
context.run(statistics_gen)
example_validator = ExampleValidator(
    statistics=statistics_gen.outputs['statistics'],
    schema=imported_baseline_schema.outputs['result']
)
context.run(example_validator)
# Show anomalies
context.show(example_validator.outputs['anomalies'])<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.08474015226746111, 'fraction_numerical': 0.007282356835484938, 'mean_word_length': 5.347338935574229, 'pattern_counts': {'":': 0, '<': 37, '<?xml version=': 0, '>': 38, 'https://': 7, 'lorem ipsum': 0, 'www.': 6, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '9', 'dir': 'jupyter-structured-clean-dedup', 'id': '89632', 'license': 'Apache-2.0', 'n_tokens_mistral': 2655, 'n_tokens_neox': 2539, 'n_words': 839, 'path': 'notebooks/tfdv-pipeline.ipynb', 'repo_name': 'MatthieuBlais/data-monitoring', 'size': '9063'}