<jupyter_start><jupyter_text>Hyperparameter TuningIn machine learning, hyperparameter tuning is the process of finding a set of optimal hyperparameters for a learning algorithm from a given hyperparameter space. Thus TFX propose a new component, with the given search space which specifies the hyperparameter configuration (name, type, range etc.). TFX will optimize the hyperparameters based on the tuning algorithm (rangom, grid, hyperband, etc..)Tuner component will utilize the KerasTuner library, all supported tuning methods will be available to TFX, including custom implementation of KerasTuner.Typical workflow of hyperparameter tuning under the KerasTuner framework:![hyperparameter tuning](../../image/hyperprameter_tuning.png)When you build a model for hypertuning, you also define the hyperparameter search space in addition to the model architecture. The model you set up for hypertuning is called a hypermodel.You can define a hypermodel through two approaches:- By using a model builder function- By subclassing the HyperModel class of the Keras Tuner API<jupyter_code>from ml_metadata.metadata_store import metadata_store
from ml_metadata.proto import metadata_store_pb2
import tensorflow as tf

base_dir = os.path.join(root_dir, Config.PIPELINE_FOLDER)
file = [i for i in os.listdir(base_dir) if 'sqlite' in i]
config = os.path.join(base_dir, file[0])

connection_config = metadata_store_pb2.ConnectionConfig()
connection_config.sqlite.filename_uri = config

store = metadata_store.MetadataStore(connection_config)
previous_execution_status = get_latest_executions(store, Config.PIPELINE_NAME, 'Transform')

if previous_execution_status:
    previous_execution_status = previous_execution_status[0].last_known_state
else:
    raise Exception('[Exception] Run the Data Ingestion Notebook before Running this...') 
    
if  previous_execution_status == 3:
    print('[INFO] previous component Execution State is Success. You can Proceed Further now..')
elif previous_execution_status == 2:
    print('[Warning] previous component Execution is in Running State')<jupyter_output>[INFO] previous component Execution State is Success. You can Proceed Further now..<jupyter_text>Defining Module FileLike the same as Trainer, Tuner component will look for a ```tuner_fn()``` function in our module file and use the function as an entry point to execute the training process.tuner_fn signature:```Input Arg:    fn_args: Holds args as name/value pairs.      - working_dir: working dir for tuning.      - train_files: List of file paths containing training tf.Example data.      - eval_files: List of file paths containing eval tf.Example data.      - train_steps: number of train steps.      - eval_steps: number of eval steps.      - schema_path: optional schema of the input data.      - transform_graph_path: optional transform graph produced by TFT.  Returns:    A namedtuple contains the following:      - tuner: A BaseTuner that will be used for tuning.      - fit_kwargs: Args to pass to tuner's run_trial function for fitting the                    model , e.g., the training and validation dataset. Required                    args depend on the above tuner's implementation.``` Most of the functions will be as same as in trainer component module file which you had runned before.<jupyter_code>%%writefile ../../../script/module.py
import kerastuner
import tensorflow as tf
import tensorflow_transform as tft
from kerastuner.engine import base_tuner
from typing import Dict, Text, Any,NamedTuple
import tensorflow_hub as hub
import sys
import os

root_dir = os.getcwd().split(os.sep)[:-3]
root_dir = '/'.join(root_dir)
sys.path.append(root_dir)

from utils.configurations.config import Config

TunerOutput = NamedTuple('TunerOutput',
                          [('tuner', base_tuner.BaseTuner),
                           ('fit_kwargs', Dict[Text, Any])])

def transformed_name(key):
    return key + '_xf'
LABEL_KEY = 'consumer_disputed'


def _gzip_reader_fn(filenames):
    return tf.data.TFRecordDataset(filenames, compression_type='GZIP')


def input_fn(file_pattern, tf_transform_output, batch_size=32):
    transformed_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=transformed_feature_spec,
        reader=_gzip_reader_fn,
        label_key=transformed_name(LABEL_KEY))
    return dataset


%%writefile ../../../script/module.py -a
def get_model(hp):
    LABEL_KEY = "consumer_disputed"
    # Loop over the features and create an input for each feature.
    # Feature name, feature dimensionality.
    ONE_HOT_FEATURES = {
    "product": 11,
    "sub_product": 45,
    "company_response": 5,
    "state": 60,
    "issue": 90
    }

    # Feature name, bucket count.
    BUCKET_FEATURES = {
    "zip_code": 10
    }

    # Feature name, value is unused.
    TEXT_FEATURES = {
    "consumer_complaint_narrative": None
    }
    # One-hot categorical features
    input_features = []
    for key, dim in ONE_HOT_FEATURES.items():
        input_features.append(
            tf.keras.Input(shape=(dim + 1,),
            name=transformed_name(key)))
        
    # Adding bucketized features
    for key, dim in BUCKET_FEATURES.items():
        input_features.append(
            tf.keras.Input(shape=(dim + 1,),
            name=transformed_name(key)))
    
    # Adding text input features
    input_texts = []
    for key in TEXT_FEATURES.keys():
        input_texts.append(
            tf.keras.Input(shape=(1,),
            name=transformed_name(key),
            dtype=tf.string))
    
    inputs = input_features + input_texts
    # Embed text features
    # Load the tf.hub module of the Universal Sentence Encoder model.
    MODULE_URL = Config.UNIVERSAL_EMBEDDING_MODEL.replace('..', root_dir)
    embed = hub.KerasLayer(MODULE_URL)
    
    reshaped_narrative = tf.reshape(input_texts[0], [-1]) # Keras inputs are two-dimensional, 
                                                            # but the encoder expects one-dimensional inputs.
    embed_narrative = embed(reshaped_narrative)
    
    deep_ff = tf.keras.layers.Reshape((512, ), input_shape=(1, 512))(embed_narrative)
    
    # for _ in range(hp.get('num_layers')):
    deep = tf.keras.layers.Dense(hp.get('units'), activation='relu')(deep_ff)
    deep = tf.keras.layers.Dense(64, activation='relu')(deep)
    deep = tf.keras.layers.Dense(16, activation='relu')(deep)
    wide_ff = tf.keras.layers.concatenate(input_features)
    wide = tf.keras.layers.Dense(hp.get('wide_units'), activation='relu')(wide_ff)
    both = tf.keras.layers.concatenate([deep, wide])
    output = tf.keras.layers.Dense(1, activation='sigmoid')(both)
    
    keras_model = tf.keras.models.Model(inputs, output)
    keras_model.compile(optimizer=tf.keras.optimizers.Adam(learning_rate=hp.get('learning_rate')),
                    loss='binary_crossentropy',
                    metrics=[
                        tf.keras.metrics.BinaryAccuracy(),
                        tf.keras.metrics.TruePositives()
                    ])
    return keras_model<jupyter_output>Appending to ../../../script/module.py<jupyter_text>we have to do some changes in ```get_model``` function. change will be like replacing the hyper-parameters with 'hp.get("Parameter_Name")' which are configured as tunable. In our case, we defined learning_rate, 1st deep layer unit and wide layer unit which will be automatically in the hyperparameter space.<jupyter_code>%%writefile ../../../script/module.py -a

# This will be called by TFX Tuner.
def tuner_fn(fn_args):
    hp = kerastuner.HyperParameters()
    # Defines search space.
    hp.Choice('wide_units', [16, 32, 64])
    hp.Int('units', min_value=256, max_value=432, step=32)
    hp.Choice('learning_rate', [1e-1, 1e-3])
    
    # RandomSearch is a subclass of Keras model Tuner.
    tuner = kerastuner.RandomSearch(
      get_model,
      max_trials=5,
      hyperparameters=hp,
      allow_new_entries=False,
      objective='val_binary_accuracy',
      directory=fn_args.working_dir,
      project_name='test')

    tf_transform_output = tft.TFTransformOutput(fn_args.transform_graph_path)
    train_dataset = input_fn(fn_args.train_files, tf_transform_output)
    eval_dataset = input_fn(fn_args.eval_files, tf_transform_output)
    return TunerOutput(
      tuner=tuner,
      fit_kwargs={'x': train_dataset,
                  'validation_data': eval_dataset,
                  'steps_per_epoch': fn_args.train_steps,
                  'validation_steps': fn_args.eval_steps})<jupyter_output>Appending to ../../../script/module.py<jupyter_text>The Tuner component makes extensive use of the Python KerasTuner API for tuning hyperparameters.Tuner takes:- tf.Examples used for training and evaluation.- A user provided module file which contains tuner_fn that defines the tuning logic, including model definition, hyperparameter search space, objective etc.- Protobuf definition of train args and eval args.- (Optional) Protobuf definition of tuning args.- (Optional) transform graph produced by an upstream Transform component.- (Optional) A data schema created by a SchemaGen pipeline component and optionally altered by the developer.With the given data, model, and objective, Tuner tunes the hyperparameters and emits the best result.<jupyter_code>%%writefile ../../../script/module.py -a

def get_serve_tf_examples_fn(model, tf_transform_output):
    model.tft_layer = tf_transform_output.transform_features_layer() # Load the preprocessing graph

    @tf.function
    def serve_tf_examples_fn(serialized_tf_examples):
        feature_spec = tf_transform_output.raw_feature_spec()
        feature_spec.pop(LABEL_KEY)
        # Parse the raw tf.Example records from the request.
        parsed_features = tf.io.parse_example(serialized_tf_examples, feature_spec)
        # Apply the preprocessing transformation to raw data.
        transformed_features = model.tft_layer(parsed_features)
        # Perform prediction with preprocessed data.
        outputs = model(transformed_features)
        return {'outputs': outputs}
    return serve_tf_examples_fn


# This will be called by TFX Generic Trainer.
def run_fn(fn_args):
    tf_transform_output = tft.TFTransformOutput(fn_args.transform_graph_path)
    train_dataset = input_fn(fn_args.train_files, tf_transform_output)
    eval_dataset = input_fn(fn_args.eval_files, tf_transform_output)
    hp = kerastuner.HyperParameters.from_config(fn_args.hyperparameters)
    model = get_model(hp)
    
    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')
    model.fit( 
        train_dataset,
        steps_per_epoch=fn_args.train_steps,
        validation_data=eval_dataset,
        validation_steps=fn_args.eval_steps,
        callbacks=[tensorboard_callback])
    
    
    signatures = {
    'serving_default':
    get_serve_tf_examples_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)
from tfx.types import channel_utils
from tfx.types import standard_artifacts
from tfx.orchestration.experimental.interactive.interactive_context import InteractiveContext
from tfx.components import Trainer
from tfx.components import Tuner
from tfx.components.base import executor_spec
from tfx.components.trainer.executor import GenericExecutor
from tfx.proto import trainer_pb2
artifacts = get_latest_artifacts(store, Config.PIPELINE_NAME, 'SchemaGen')
example_schema = find_latest_artifacts_by_type(store, artifacts, standard_artifacts.Schema.TYPE_NAME)
example_schema = channel_utils.as_channel(example_schema)
artifacts = get_latest_artifacts(store, Config.PIPELINE_NAME, 'Transform')
transform_examples = find_latest_artifacts_by_type(store, artifacts, standard_artifacts.Examples.TYPE_NAME)
transform_examples = channel_utils.as_channel(transform_examples)

transform_graph = find_latest_artifacts_by_type(store, artifacts, standard_artifacts.TransformGraph.TYPE_NAME)
transform_graph = channel_utils.as_channel(transform_graph)
TRAINING_STEPS = 1000
EVALUATION_STEPS = 100
module_file = '../../../script/module.py'

tuner = Tuner(
    module_file=module_file,
    schema=example_schema,
    examples=transform_examples,
    transform_graph=transform_graph,
    train_args=trainer_pb2.TrainArgs(num_steps=EVALUATION_STEPS),
    eval_args=trainer_pb2.EvalArgs(num_steps=EVALUATION_STEPS))

trainer = Trainer(
    module_file=module_file,
    custom_executor_spec=executor_spec.ExecutorClassSpec(GenericExecutor),
    examples=transform_examples,
    transform_graph=transform_graph,
    schema=example_schema,

    hyperparameters=tuner.outputs['best_hyperparameters'],
    train_args=trainer_pb2.TrainArgs(num_steps=TRAINING_STEPS),
    eval_args=trainer_pb2.EvalArgs(num_steps=EVALUATION_STEPS))<jupyter_output>WARNING:absl:From <ipython-input-12-7192c46db06f>:15: The name tfx.components.base.executor_spec.ExecutorClassSpec is deprecated. Please use tfx.dsl.components.base.executor_spec.ExecutorClassSpec instead.<jupyter_text>The output artifact of the tuner component will be used  by trainer componet which will be used to train model with the best hyper-parameters searched by tuner<jupyter_code>pipeline_name = Config.PIPELINE_NAME
pipeline_root = os.path.join(root_dir, f'temp_')
beam_args = [
    '--runner=DirectRunner'
]

if not os.path.exists(pipeline_root):
    raise Exception('Please do follow the notebook sequence')

context = InteractiveContext(pipeline_name = pipeline_name,
                            pipeline_root = pipeline_root,
                            beam_pipeline_args = beam_args)

context.run(tuner)
context.run(trainer)<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.08626697958893394, 'fraction_numerical': 0.005760614465542991, 'mean_word_length': 3.960141093474427, 'pattern_counts': {'":': 7, '<': 18, '<?xml version=': 0, '>': 18, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 3, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '5', 'dir': 'jupyter-structured-clean-dedup', 'id': '93046', 'license': 'Apache-2.0', 'n_tokens_mistral': 4284, 'n_tokens_neox': 4134, 'n_words': 1049, 'path': 'notebooks/add-ons/Model Training/Tuner + Traning (Hyperparameter tunning).ipynb', 'repo_name': 'jagan-mathematics/Tensorflow-Extended-tutorial', 'size': '14061'}