Cuando estamos trabajando con machine learning, después de entrenar nuestros modelos podemos hacer predicciones en batch sobre texto en nuestros ejemplos de test en local. En este artículo, veremos cómo hacer esto con un ejercicio de análisis de sentimientos con un dataset de Twitter.
Predicciones en batch sobre texto
Para hacer predicciones en batch sobre texto tenemos que hacer dos procesos: uno es el de las validaciones locales y el otro tiene que ver con el proceso dentro de Google Cloud Platform.
Generar el predict en Google Cloud Platform
Lo primero que tendremos que hacer para realizar predicciones en batch sobre texto es generar el predict. Esto se hace creando un script de nombre predict.py, de modo que allí pondremos toda la información del algoritmo que se ejecutará internamente.
Veamos:
#Generar predicciones en batch sobre texto
class Predict (beam.DoFn):
def __init__ (
self, model_dir,
):
self.model_dir = model_dir
self.model = None
self.tokenizer = None
def setup (self):
keras_model_path = os.path.join (self.model_dir, KERAS_MODEL)
with tempfile.NamedTemporary File (suffix = ".h5") as local_file:
with tf.io.gfile.GFile (keras_model_path, mode = "rb") as gcs:file:
local_file.write (gcs_file.read ())
self.model = tf.keras.models. load_model (local_file.name)
def decode_sentiment (self, score, include_neutral = True):
if include_neutral:
label = NEUTRAL
if score = SENTIMENT_THRESHOLD [0]:
label = NEGATIVE
elif score <= SENTIMENT_THRESHOLD [1]:
label = POSITIVE
return label
else:
return NEGATIVE if score < 0.5 else POSITIVE
#Generar predicciones en batch sobre texto
def process (self, element):
start_at = time.time ()
#Tokenize
x_test = pad_sequences (
self.tokenizer.texts_to_sequences ([element]), maxlen = SEQUENCE_LENGTH
)
#Predict
score = self.model.predict ([x_test]) [0]
#Decode sentiment
label = self.decode_sentiment (score)
yield {
"text": element,
"label": label,
"score": float (score),
"elapsed_time": time.time () - start_at,
}
def __init__ (self, encoding: str):
#latin - 1
#iso - 8859 - 1
self.enconding = encoding
def encode (self, value):
return value.encode (self.enconding)
def decode (self, value):
return value.decode (self.enconding)
def is:deterministic (self):
return True
def run (model_dir, source, sink, beam_options = None):
with beam.Pipeline (options = beam_options) as p:
__ = (
p
| "Read data" >> source
# | "Preprocess" >> beam.ParDo (PreprocessTextFn (model_dir, "ID"))
| "Predict" >> beam.ParDo (Predict (model_dir))
| "Format as JSON" >> beam.Map (json.dumps)
| "Write predictions" >> sink
)
#Generar predicciones en batch sobre texto
if __name__ == "__main__":
" " " Main function " " "
parser = argparse.ArgumentParser (
formatter_class = argparse.ArgumentDefaultsHelpFormatter
)
parser.add_argument (
"--work - dir",
dest = "work_dir",
required = True,
help = "Directory for temporary files and preprocessed dataset to. "
"This can be a Google Cloud Storage path. ",
)
parser.add_argument (
"--model - dir",
dest = "model_dir",
required = True,
help = "Path to the exported TensorFlow model. "
"This can be a Google Cloud Storage path. ",
)
verbs = parser.add:subparsers (dest = "verb")
batch_verb = verbs.add_subparsers (dest = "verb")
batch_verb.add_argument (
"--inputs- dir",
dest = "inputs_dir",
required = True,
help = "Input directory where CSV data files are read from."
"This can be a Google Cloud Storage path. ",
)
batch_verb.add_argument (
"--outputs- dir",
dest = "outputs_dir",
required = True,
help = "Directory to store prediction results. "
"This can be a Google Cloud Storage path. ",
)
args, pipeline_args = parser.parse_known_args ()
print (args)
beam_options = PipelineOptions (pipeline_args)
beam_options.view_as (SetupOptions).save_main _session = True
#beam_options.view_as (DirectOptions).direct_ num_workers = 0
#Generar predicciones en batch sobre texto
project = beam_options.view_as (GoogleCloudOptions).project
if args.verb == "batch":
results_prefix = os.path.join (args.outputs_dir, "part")
source = ReadFromText (args.inputs_dir, coder = CustomCoder ("latin - 1"))
sink = WriteToText (results_prefix)
else:
parser.print_usage ()
sys.exit (1)
run (args.model_dir, source, sink, beam_options)Aquí tenemos, entonces, el if name, el parser de argumentos y los parámetros, el work dir, el model dir. Como en este caso vamos a hacer una predicción, vamos a hacer el model dir para poder decirle al algoritmo dónde está alojado el path con el que vamos a entrenar.
También tenemos las entradas y las salidas.
Realizamos una condicional en el que, si es batch, nos va a generar dos transformaciones iniciales que van a ser la fuente y el lugar en el que vamos a dejar los datos. La fuente va a estar relacionada con la lectura de un fichero y donde lo vamos a volcar es la escritura del texto.
Validación local
Lo único que queda para realizar predicciones en batch sobre texto es ejecutar la siguiente celda:
#Generar predicciones en batch sobre texto
from datetime import datetime
#current date and time
TIMESTAMP = datetime.now ().strftime ('%Y - %m - %d_ %H - %M - %S')#Generar predicciones en batch sobre texto
! python3 predict.py \
--work - dir $WORK_DIR \
--MODEL- dir $WORK_DIR/data/model \
batch \
--inputs - dir $WORK_DIR/data/transformed_data/test/part* \
--outputs- dir $WORK_DIR/predictions/$TIMESTAMP
¿Qué sigue?
Para seguir formándote en alguna de las numerosas temáticas del mundo del Big Data, puedes acudir a nuestros cursos y bootcamps. En KeepCoding, expertos en el sector te guiarán a través de la teoría y la práctica para que, en unos pocos meses, te conviertas en un gran profesional IT. Échale un ojo al temario de nuestro Big Data, Inteligencia Artificial & Machine Learning Full Stack Bootcamp y descubre esta formación intensiva e íntegra. ¡Solicita ahora mismo más información y da el salto que cambiará tu futuro!
