machine_learning_pipeline.py

# -*- coding: utf-8 -*-
"""machine_learning_pipeline.ipynb

Automatically generated by Colaboratory.

Original file is located at
    https://colab.research.google.com/drive/1rswX2J31jv_MOOfILZ5EdcaSuQCjTb05

## **Install and Import Library**
"""

!pip install tfx tensorflow_model_analysis

import os
import pandas as pd
import tensorflow_model_analysis as tfma
import zipfile
from tfx.components import CsvExampleGen, StatisticsGen, SchemaGen, ExampleValidator
from tfx.components import Transform, Trainer, Tuner, Evaluator, Pusher
from tfx.proto import example_gen_pb2, trainer_pb2, pusher_pb2
from tfx.orchestration.experimental.interactive.interactive_context import InteractiveContext
from tfx.dsl.components.common.resolver import Resolver
from tfx.dsl.input_resolution.strategies.latest_blessed_model_strategy import LatestBlessedModelStrategy
from tfx.types import Channel
from tfx.types.standard_artifacts import Model, ModelBlessing
from google.colab import files

"""## **Set Variable of Pipeline Directory**"""

PIPELINE_NAME = "anashaneef-pipeline"
SCHEMA_PIPELINE_NAME = "anashaneef-tfdv-schema"

PIPELINE_ROOT = os.path.join('pipelines', PIPELINE_NAME)
METADATA_PATH = os.path.join('metadata', PIPELINE_NAME, 'metadata.db')
SERVING_MODEL_DIR = os.path.join('serving_model', PIPELINE_NAME)

DATA_ROOT = "data"

interactive_context = InteractiveContext(pipeline_root=PIPELINE_ROOT)

"""## **Dataframe Information**

**Variables in dataset**

* Headline : Headline/title
* Clickbait : Information whether including clickbait or not
"""

clickbait = pd.read_csv("data/clickbait.csv")
clickbait.head()

"""## **Data Ingestion Phase**"""

# Use Example Gen
output = example_gen_pb2.Output(
    split_config=example_gen_pb2.SplitConfig(splits=[
        example_gen_pb2.SplitConfig.Split(name="train", hash_buckets=9),
        example_gen_pb2.SplitConfig.Split(name="eval", hash_buckets=1),
    ])
)

example_gen = CsvExampleGen(input_base=DATA_ROOT, output_config=output)
interactive_context.run(example_gen)

"""## **Data Validation Phase**

**Summary Statistic**
"""

# Use Statistic Gen
statistics_gen = StatisticsGen(
    examples=example_gen.outputs["examples"]
)
interactive_context.run(statistics_gen)

interactive_context.show(statistics_gen.outputs["statistics"])

"""**Data Schema**"""

# Use Schema Gen
schema_gen = SchemaGen(
    statistics=statistics_gen.outputs["statistics"]
)
interactive_context.run(schema_gen)

interactive_context.show(schema_gen.outputs["schema"])

"""**Identifying anomalies in the dataset**"""

# Use example Validator
example_validator = ExampleValidator(
    statistics=statistics_gen.outputs["statistics"],
    schema=schema_gen.outputs["schema"],
)
interactive_context.run(example_validator)

interactive_context.show(example_validator.outputs['anomalies'])

"""## **Data Preprocessing**"""

TRANSFORM_MODULE_FILE = "clickbait_transform.py"

# Commented out IPython magic to ensure Python compatibility.
# %%writefile {TRANSFORM_MODULE_FILE}
# import string
# import tensorflow as tf
# 
# LABEL_KEY = "clickbait"
# FEATURE_KEY = "headline"
# 
# def transformed_name(key):
#     return key + "_xf"
# 
# def preprocessing_fn(inputs):
#     outputs = {}
#     outputs[transformed_name(FEATURE_KEY)] = tf.strings.lower(
#         inputs[FEATURE_KEY]
#     )
#     outputs[transformed_name(LABEL_KEY)] = tf.cast(
#         inputs[LABEL_KEY], tf.int64
#     )
# 
#     return outputs

transform = Transform(
    examples=example_gen.outputs["examples"],
    schema=schema_gen.outputs["schema"],
    module_file=os.path.abspath(TRANSFORM_MODULE_FILE)
)
interactive_context.run(transform)

"""## **Model Development Phase**

**Hyperparameter Tuning**
"""

TUNER_MODULE_FILE = "clickbait_tuner.py"

# Commented out IPython magic to ensure Python compatibility.
# %%writefile {TUNER_MODULE_FILE}
# import keras_tuner as kt
# import tensorflow as tf
# import tensorflow_transform as tft
# from typing import NamedTuple, Dict, Text, Any
# from keras_tuner.engine import base_tuner
# from tensorflow.keras import layers
# from tfx.components.trainer.fn_args_utils import FnArgs
# 
# # Tuner parameters
# LABEL_KEY = "clickbait"
# FEATURE_KEY = "headline"
# NUM_EPOCHS = 5
# 
# TunerFnResult = NamedTuple("TunerFnResult", [
#     ("tuner", base_tuner.BaseTuner),
#     ("fit_kwargs", Dict[Text, Any]),
# ])
# 
# early_stopping_callback = tf.keras.callbacks.EarlyStopping(
#     monitor="val_binary_accuracy",
#     mode="max",
#     verbose=1,
#     patience=10,
# )
# 
# # transformed_name function
# # used to rename features that have gone through the transform process.
# def transformed_name(key):
#     return key + "_xf"
# 
# # gzip_reader_fn
# # used to load data in TFRecord format.
# def gzip_reader_fn(filenames):
#     return tf.data.TFRecordDataset(filenames, compression_type="GZIP")
# 
# # input_fn
# # used to load the transformed_feature generated by the Transform component and split it into batches.
# def input_fn(file_pattern, tf_transform_output, num_epochs, batch_size=64):
#     transform_feature_spec = (
#         tf_transform_output.transformed_feature_spec().copy()
#     )
#     
#     dataset = tf.data.experimental.make_batched_features_dataset(
#         file_pattern=file_pattern,
#         batch_size=batch_size,
#         features=transform_feature_spec,
#         reader=gzip_reader_fn,
#         num_epochs=num_epochs,
#         label_key=transformed_name(LABEL_KEY),
#     )
#     
#     return dataset
# 
# # model_builder function
# def model_builder(hp, vectorizer_layer):
#     num_hidden_layers = hp.Choice(
#         "num_hidden_layers", values=[1, 2]
#     )
#     embed_dims = hp.Int(
#         "embed_dims", min_value=16, max_value=128, step=32
#     )
#     lstm_units= hp.Int(
#         "lstm_units", min_value=32, max_value=128, step=32
#     )
#     dense_units = hp.Int(
#         "dense_units", min_value=32, max_value=256, step=32
#     )
#     dropout_rate = hp.Float(
#         "dropout_rate", min_value=0.1, max_value=0.5, step=0.1
#     )
#     learning_rate = hp.Choice(
#         "learning_rate", values=[1e-2, 1e-3, 1e-4]
#     )
# 
#     inputs = tf.keras.Input(
#         shape=(1,), name=transformed_name(FEATURE_KEY), dtype=tf.string
#     )
#     
#     x = vectorizer_layer(inputs)
#     x = layers.Embedding(input_dim=5000, output_dim=embed_dims)(x)
#     x = layers.Bidirectional(layers.LSTM(lstm_units))(x)
# 
#     for _ in range(num_hidden_layers):
#         x = layers.Dense(dense_units, activation=tf.nn.relu)(x)
#         x = layers.Dropout(dropout_rate)(x)
#     
#     outputs = layers.Dense(1, activation=tf.nn.sigmoid)(x)
#     
#     model = tf.keras.Model(inputs=inputs, outputs=outputs)
# 
#     model.compile(
#         optimizer=tf.keras.optimizers.Adam(learning_rate=learning_rate),
#         loss=tf.keras.losses.BinaryCrossentropy(from_logits=True),
#         metrics=["binary_accuracy"],
#     )
#     
#     return model
# 
# # tuner_fn
# def tuner_fn(fn_args: FnArgs):
#     tf_transform_output = tft.TFTransformOutput(fn_args.transform_graph_path)
#     
#     train_set = input_fn(
#         fn_args.train_files[0], tf_transform_output, NUM_EPOCHS
#     )
#     eval_set = input_fn(
#         fn_args.eval_files[0], tf_transform_output, NUM_EPOCHS
#     )
# 
#     vectorizer_dataset = train_set.map(
#         lambda f, l: f[transformed_name(FEATURE_KEY)]
#     )
# 
#     vectorizer_layer = layers.TextVectorization(
#         max_tokens=5000,
#         output_mode="int",
#         output_sequence_length=500,
#     )
#     vectorizer_layer.adapt(vectorizer_dataset)
#     
#     tuner = kt.Hyperband(
#         hypermodel=lambda hp: model_builder(hp, vectorizer_layer),
#         objective=kt.Objective('binary_accuracy', direction='max'),
#         max_epochs=NUM_EPOCHS,
#         factor=3,
#         directory=fn_args.working_dir,
#         project_name="kt_hyperband",
#     )
#     
#     return TunerFnResult(
#         tuner=tuner,
#         fit_kwargs={
#             "callbacks": [early_stopping_callback],
#             "x": train_set,
#             "validation_data": eval_set,
#             "steps_per_epoch": fn_args.train_steps,
#             "validation_steps": fn_args.eval_steps,
#         },
#     )

"""**tuner**"""

tuner = Tuner(
    module_file=os.path.abspath(TUNER_MODULE_FILE),
    examples=transform.outputs["transformed_examples"],
    transform_graph=transform.outputs["transform_graph"],
    schema=schema_gen.outputs["schema"],
    train_args=trainer_pb2.TrainArgs(splits=["train"], num_steps=800),
    eval_args=trainer_pb2.EvalArgs(splits=["eval"], num_steps=400),
)
interactive_context.run(tuner)

"""## **Model Training**"""

TRAINER_MODULE_FILE = "clickbait_trainer.py"

# Commented out IPython magic to ensure Python compatibility.
# %%writefile {TRAINER_MODULE_FILE}
# import os
# import tensorflow as tf
# import tensorflow_transform as tft
# import tensorflow_hub as hub
# from tensorflow.keras import layers
# from tfx.components.trainer.fn_args_utils import FnArgs
# 
# 
# LABEL_KEY = "clickbait"
# FEATURE_KEY = "headline"
# 
# 
# def transformed_name(key): 
#     return f"{key}_xf"
# 
# 
# def gzip_reader_fn(filenames):
#     return tf.data.TFRecordDataset(filenames, compression_type="GZIP")
# 
# 
# def input_fn(file_pattern, tf_transform_output, num_epochs, batch_size=64):
#     transform_feature_spec = (
#         tf_transform_output.transformed_feature_spec().copy()
#     )
#     
#     dataset = tf.data.experimental.make_batched_features_dataset(
#         file_pattern=file_pattern,
#         batch_size=batch_size,
#         features=transform_feature_spec,
#         reader=gzip_reader_fn,
#         num_epochs=num_epochs,
#         label_key=transformed_name(LABEL_KEY),
#     )
#     
#     return dataset
# 
# 
# def model_builder(vectorizer_layer, hp):
#     inputs = tf.keras.Input(
#         shape=(1,), name=transformed_name(FEATURE_KEY), dtype=tf.string
#     )
# 
#     x = vectorizer_layer(inputs)
#     x = layers.Embedding(input_dim=5000, output_dim=hp["embed_dims"])(x)
#     x = layers.Bidirectional(layers.LSTM(hp["lstm_units"]))(x)
# 
#     for _ in range(hp["num_hidden_layers"]):
#         x = layers.Dense(hp["dense_units"], activation=tf.nn.relu)(x)
#         x = layers.Dropout(hp["dropout_rate"])(x)
# 
#     outputs = layers.Dense(1, activation=tf.nn.sigmoid)(x)
#     
#     model = tf.keras.Model(inputs=inputs, outputs = outputs)
# 
#     model.compile(
#         optimizer=tf.keras.optimizers.Adam(learning_rate=hp["learning_rate"]),
#         loss=tf.keras.losses.BinaryCrossentropy(),
#         metrics=[tf.keras.metrics.BinaryAccuracy()],
#     )
#     
#     model.summary()
#     
#     return model
# 
# 
# def _get_serve_tf_example_fn(model, tf_transform_output):
#     model.tft_layer = tf_transform_output.transform_features_layer()
#     
#     @tf.function
#     def serve_tf_examples_fn(serialized_tf_examples):
#         feature_spec = tf_transform_output.raw_feature_spec()
#         feature_spec.pop(LABEL_KEY)
#         
#         parsed_features = tf.io.parse_example(
#             serialized_tf_examples, feature_spec
#         )
#         transformed_features = model.tft_layer(parsed_features)
#         
#         # get predictions using transformed features
#         return model(transformed_features)
#     
#     return serve_tf_examples_fn
# 
# 
# def run_fn(fn_args: FnArgs):
#     hp = fn_args.hyperparameters["values"]
#     log_dir = os.path.join(os.path.dirname(fn_args.serving_model_dir), "logs")
#     
#     tensorboard_callback = tf.keras.callbacks.TensorBoard(
#         log_dir=log_dir, update_freq="batch"
#     )
#     early_stopping_callback = tf.keras.callbacks.EarlyStopping(
#         monitor="val_binary_accuracy",
#         mode="max",
#         verbose=1,
#         patience=10,
#     )
#     model_checkpoint_callback = tf.keras.callbacks.ModelCheckpoint(
#         fn_args.serving_model_dir,
#         monitor="val_binary_accuracy",
#         mode="max",
#         verbose=1,
#         save_best_only=True,
#     )
#     callbacks = [
#         tensorboard_callback, 
#         early_stopping_callback, 
#         model_checkpoint_callback
#     ]
#     
#     tf_transform_output = tft.TFTransformOutput(fn_args.transform_graph_path)
#     
#     train_set = input_fn(
#         fn_args.train_files, tf_transform_output, hp["tuner/epochs"]
#     )
#     eval_set = input_fn(
#         fn_args.eval_files, tf_transform_output, hp["tuner/epochs"]
#     )
#     
#     vectorizer_dataset = train_set.map(
#         lambda f, l: f[transformed_name(FEATURE_KEY)]
#     )
#     
#     vectorizer_layer = layers.TextVectorization(
#         max_tokens=5000,
#         output_mode="int",
#         output_sequence_length=500,
#     )
#     vectorizer_layer.adapt(vectorizer_dataset)
# 
#     model = model_builder(vectorizer_layer, hp)
#     
#     model.fit(
#         x=train_set,
#         steps_per_epoch=fn_args.train_steps,
#         validation_data=eval_set,
#         validation_steps=fn_args.eval_steps,
#         callbacks=callbacks,
#         epochs=hp["tuner/epochs"],
#         verbose=1,
#     )
#     
#     signatures = {
#         "serving_default": _get_serve_tf_example_fn(
#             model, tf_transform_output
#         ).get_concrete_function(
#             tf.TensorSpec(
#                 shape=[None],
#                 dtype=tf.string,
#                 name="examples",
#             )
#         )
#     }
#     
#     model.save(
#         fn_args.serving_model_dir, 
#         save_format="tf", 
#         signatures=signatures
#     )

trainer = Trainer(
    module_file=os.path.abspath(TRAINER_MODULE_FILE),
    examples=transform.outputs["transformed_examples"],
    transform_graph=transform.outputs["transform_graph"],
    schema=schema_gen.outputs["schema"],
    hyperparameters=tuner.outputs['best_hyperparameters'],
    train_args=trainer_pb2.TrainArgs(splits=["train"], num_steps=800),
    eval_args=trainer_pb2.EvalArgs(splits=["eval"], num_steps=400),
)
interactive_context.run(trainer)

"""## **Analysis and Model Validation Phase**

**Resolver Component**
"""

model_resolver = Resolver(
    strategy_class=LatestBlessedModelStrategy,
    model=Channel(type=Model),
    model_blessing=Channel(type=ModelBlessing),
).with_id("Latest_blessed_model_resolver")

interactive_context.run(model_resolver)

"""**Evaluator Component**"""

import tensorflow_model_analysis as tfma 
 
eval_config = tfma.EvalConfig(
    model_specs=[tfma.ModelSpec(label_key='clickbait')],
    slicing_specs=[tfma.SlicingSpec()],
    metrics_specs=[
        tfma.MetricsSpec(metrics=[
            
            tfma.MetricConfig(class_name='ExampleCount'),
            tfma.MetricConfig(class_name='AUC'),
            tfma.MetricConfig(class_name='FalsePositives'),
            tfma.MetricConfig(class_name='TruePositives'),
            tfma.MetricConfig(class_name='FalseNegatives'),
            tfma.MetricConfig(class_name='TrueNegatives'),
            tfma.MetricConfig(class_name='BinaryAccuracy',
                threshold=tfma.MetricThreshold(
                    value_threshold=tfma.GenericValueThreshold(
                        lower_bound={'value':0.5}),
                    change_threshold=tfma.GenericChangeThreshold(
                        direction=tfma.MetricDirection.HIGHER_IS_BETTER,
                        absolute={'value':0.0001})
                    )
            )
        ])
    ]
 
)

evaluator = Evaluator(
    examples=example_gen.outputs["examples"],
    model=trainer.outputs["model"],
    baseline_model=model_resolver.outputs["model"],
    eval_config=eval_config,
)
interactive_context.run(evaluator)

"""**Visualize the evaluation results**"""

eval_result = evaluator.outputs['evaluation'].get()[0].uri
tfma_result = tfma.load_eval_result(eval_result)
tfma.view.render_slicing_metrics(tfma_result)
tfma.addons.fairness.view.widget_view.render_fairness_indicator(
    tfma_result
    )

"""## **Pusher Component**"""

pusher = Pusher(
    model=trainer.outputs["model"],
    model_blessing=evaluator.outputs["blessing"],
    push_destination=pusher_pb2.PushDestination(
        filesystem=pusher_pb2.PushDestination.Filesystem(
            base_directory=os.path.join(
                SERVING_MODEL_DIR, "clickbait-model"
            ),
        )
    )
)

interactive_context.run(pusher)