<jupyter_start><jupyter_text>Traininig the Inclusive classifier with tf.Keras using data in TFRecord format**tf.keras Inclusive classifier** This notebooks trains a neural network for the particle classifier using the Inclusive Classifier, using as input the list of recunstructed particles with the low level features + the high level features. Data is prepared from Parquet using Apache Spark, and written into TFRecord format. Data in TFRecord format is read from TensorFlow using tf.data and tf.io in tf.keras.To run this notebook we used the following configuration:* *Software stack*: TensorFlow 2.0.1* *Platform*: CentOS 7, Python 3.6<jupyter_code>import tensorflow as tf
import numpy as np
from tensorflow.keras.optimizers import Adam
from tensorflow.keras import Sequential, Input, Model
from tensorflow.keras.layers import Masking, Dense, Activation, GRU, Dropout, concatenate

tf.version.VERSION
# only needed for TensorFlow 1.x
# tf.enable_eager_execution()<jupyter_output><empty_output><jupyter_text>Configure distributed training using tf.distributeThis notebook shows an example of distributed training with tf.keras using 4 concurrent executions on a single machine.The test machine has 24 physical cores it has been notes that a serial execution of the training would leave spare capacity. With distributed training we can "use all the CPU on the box". - TensorFlow MultiWorkerMirroredStrategy is used to distribute the training.- Configuration of the workers is done using the OS enviroment variable **TF_CONFIG**.- **nodes_endpoints** configures the list of machines and ports that will be used. In this example, we use 3 workers on the same machines, you can use this to distribute over multiple machines too- **worker_number** will be unique for each worker, numbering starts from 0- Worker number 0 will be the master. - You need to run the 4 notebooks for the 4 configured workers at the same time (training will only start when all 4 workers are active)<jupyter_code># Each worker will have a unique worker_number, numbering starts from 0
worker_number=1

nodes_endpoints = ["localhost:12345", "localhost:12346", "localhost:12347", "localhost:12348"]
number_workers = len(nodes_endpoints)

import os
import json
os.environ['TF_CONFIG'] = json.dumps({
    'cluster': {
        'worker': nodes_endpoints
    },
    'task': {'type': 'worker', 'index': worker_number}
})

strategy = tf.distribute.experimental.MultiWorkerMirroredStrategy()<jupyter_output><empty_output><jupyter_text>Create the Keras model for the inclusive classifier hooking with tf.distribute<jupyter_code># This implements the distributed stratedy for model
with strategy.scope():
    ## GRU branch
    gru_input = Input(shape=(801,19), name='gru_input')
    a = gru_input
    a = Masking(mask_value=0.)(a)
    a = GRU(units=50,activation='tanh')(a)
    gruBranch = Dropout(0.2)(a)
    
    hlf_input = Input(shape=(14), name='hlf_input')
    b = hlf_input
    hlfBranch = Dropout(0.2)(b)

    c = concatenate([gruBranch, hlfBranch])
    c = Dense(25, activation='relu')(c)
    output = Dense(3, activation='softmax')(c)
    
    model = Model(inputs=[gru_input, hlf_input], outputs=output)
    
    ## Compile model
    optimizer = 'Adam'
    loss = 'categorical_crossentropy'
    model.compile(loss=loss, optimizer=optimizer, metrics=["accuracy"] )
model.summary()<jupyter_output>Model: "model"
__________________________________________________________________________________________________
Layer (type)                    Output Shape         Param #     Connected to                     
==================================================================================================
gru_input (InputLayer)          [(None, 801, 19)]    0                                            
__________________________________________________________________________________________________
masking (Masking)               (None, 801, 19)      0           gru_input[0][0]                  
__________________________________________________________________________________________________
gru (GRU)                       (None, 50)           10650       masking[0][0]                    
__________________________________________________________________________________________________
hlf_input (InputLayer)          [(None, 14)]         0                                        [...]<jupyter_text>Load test and training data in TFRecord format, using tf.data and tf.io<jupyter_code>PATH = "/local3/lucatests/Data/"

# test dataset 
files_test_dataset = tf.data.Dataset.list_files(PATH + "testUndersampled.tfrecord/part-r*", shuffle=False)

# training dataset 
files_train_dataset = tf.data.Dataset.list_files(PATH + "trainUndersampled.tfrecord/part-r*", seed=4242)
# tunable
num_parallel_reads=16

test_dataset = files_test_dataset.prefetch(buffer_size=tf.data.experimental.AUTOTUNE).interleave(
    tf.data.TFRecordDataset, 
    cycle_length=num_parallel_reads,
    num_parallel_calls=tf.data.experimental.AUTOTUNE)

train_dataset = files_train_dataset.prefetch(buffer_size=tf.data.experimental.AUTOTUNE).interleave(
    tf.data.TFRecordDataset, cycle_length=num_parallel_reads,
    num_parallel_calls=tf.data.experimental.AUTOTUNE)
# Function to decode TF records into the required features and labels
def decode(serialized_example):
    deser_features = tf.io.parse_single_example(
      serialized_example,
      # Defaults are not specified since both keys are required.
      features={
          'HLF_input': tf.io.FixedLenFeature((14), tf.float32),
          'GRU_input': tf.io.FixedLenFeature((801,19), tf.float32),
          'encoded_label': tf.io.FixedLenFeature((3), tf.float32),
          })
    return((deser_features['GRU_input'], deser_features['HLF_input']), deser_features['encoded_label'])
# use for debugging
# for record in test_dataset.take(1):
#     print(record)
parsed_test_dataset=test_dataset.map(decode, num_parallel_calls=tf.data.experimental.AUTOTUNE)
parsed_train_dataset=train_dataset.map(decode, num_parallel_calls=tf.data.experimental.AUTOTUNE).cache()
# use for debugging
# Show and example of the parsed data
# for record in parsed_test_dataset.take(1):
#    print(record)
# tunable
batch_size = 128 * number_workers

train=parsed_train_dataset.batch(batch_size)
train=train.prefetch(buffer_size=tf.data.experimental.AUTOTUNE)
train=train.repeat()
train
num_train_samples=3426083   # there are 3426083 samples in the training dataset

steps_per_epoch=num_train_samples//batch_size
steps_per_epoch
# tunable
test_batch_size = 1024 

test=parsed_test_dataset.batch(test_batch_size)
test=test.prefetch(buffer_size=tf.data.experimental.AUTOTUNE)
test=test.repeat()
num_test_samples=856090 # there are 856090 samples in the test dataset

validation_steps=num_test_samples//test_batch_size  
validation_steps<jupyter_output><empty_output><jupyter_text>Train the tf.keras model<jupyter_code># train the Keras model

# tunable
num_epochs = 6

# callbacks = [ tf.keras.callbacks.TensorBoard(log_dir='./logs') ]
callbacks = []
    
%time history = model.fit(train, steps_per_epoch=steps_per_epoch, \
                          epochs=num_epochs, callbacks=callbacks, verbose=1)
                          #validation_data=test, validation_steps=validation_steps, \

PATH="/local1/lucatests/SparkDLTrigger/Training_TFKeras_Distributed/model/"
model.save(PATH + "mymodel" + str(worker_number) + ".h5", save_format='h5')<jupyter_output><empty_output><jupyter_text>Training history performance metrics<jupyter_code>%matplotlib notebook
import matplotlib.pyplot as plt 
plt.style.use('seaborn-darkgrid')
# Graph with loss vs. epoch

plt.figure()
plt.plot(history.history['loss'], label='train')
plt.plot(history.history['val_loss'], label='validation')
plt.ylabel('loss')
plt.xlabel('epoch')
plt.legend(loc='upper right')
plt.title("HLF classifier loss")
plt.show()
# Graph with accuracy vs. epoch
%matplotlib notebook
plt.figure()
plt.plot(history.history['accuracy'], label='train')
plt.plot(history.history['val_accuracy'], label='validation')
plt.ylabel('Accuracy')
plt.xlabel('epoch')
plt.legend(loc='lower right')
plt.title("HLF classifier accuracy")
plt.show()<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.10682095006090134, 'fraction_numerical': 0.017295980511571253, 'mean_word_length': 4.819277108433735, 'pattern_counts': {'":': 0, '<': 24, '<?xml version=': 0, '>': 24, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '6', 'dir': 'jupyter-structured-clean-dedup', 'id': '161037', 'license': 'Apache-2.0', 'n_tokens_mistral': 2496, 'n_tokens_neox': 2358, 'n_words': 614, 'path': 'Training_TFKeras_CPU_Distributed/MultiWorker_Notebooks/4.3a-Worker1_Training-InclusiveClassifier-TF_Keras_TFRecord.ipynb', 'repo_name': 'cerndb/SparkDLTrigger', 'size': '8210'}