Transformer model sentiment analysis example (#36)

Author: 1vn

docs/applications/implementations/aggregators.md

@@ -42,7 +42,7 @@ def aggregate_spark(data, columns, args):
 The following packages have been pre-installed and can be used in your implementations:
 
 ```text
-pyspark==2.4.0
+pyspark==2.4.1
 boto3==1.9.78
 msgpack==0.6.1
 numpy>=1.13.3,<2

docs/applications/implementations/transformers.md

@@ -86,7 +86,7 @@ def reverse_transform_python(transformed_value, args):
 The following packages have been pre-installed and can be used in your implementations:
 
 ```text
-pyspark==2.4.0
+pyspark==2.4.1
 boto3==1.9.78
 msgpack==0.6.1
 numpy>=1.13.3,<2

docs/applications/resources/environments.md

@@ -35,7 +35,7 @@ data:
 
 #### CSV Config
 
-To help ingest different styles of CSV files, Cortex supports the parameters listed below. All of these parameters are optional. A description and default values for each parameter can be found in the [PySpark CSV Documentation](https://spark.apache.org/docs/2.4.0/api/python/pyspark.sql.html#pyspark.sql.DataFrameReader.csv).
+To help ingest different styles of CSV files, Cortex supports the parameters listed below. All of these parameters are optional. A description and default values for each parameter can be found in the [PySpark CSV Documentation](https://spark.apache.org/docs/2.4.1/api/python/pyspark.sql.html#pyspark.sql.DataFrameReader.csv).
 
 ```yaml
 csv_config:

examples/mnist/implementations/models/t2t.py

@@ -35,6 +35,6 @@ def transform_tensorflow(features, labels, model_config):
     features["inputs"] = tf.reshape(features["image_pixels"], hparams["input_shape"])
 
     # t2t expects this key and dimensionality
-    features["targets"] = tf.expand_dims(labels, 0)
+    features["targets"] = tf.expand_dims(tf.expand_dims(labels, -1), -1)
 
     return features, labels

examples/reviews/implementations/aggregators/max_length.py

@@ -1,15 +1,13 @@
 def aggregate_spark(data, columns, args):
-    from pyspark.ml.feature import StopWordsRemover, RegexTokenizer
+    from pyspark.ml.feature import RegexTokenizer
     import pyspark.sql.functions as F
     from pyspark.sql.types import IntegerType
 
     regexTokenizer = RegexTokenizer(inputCol=columns["col"], outputCol="token_list", pattern="\\W")
     regexTokenized = regexTokenizer.transform(data)
 
-    remover = StopWordsRemover(inputCol="token_list", outputCol="filtered_word_list")
     max_review_length_row = (
-        remover.transform(regexTokenized)
-        .select(F.size(F.col("filtered_word_list")).alias("word_count"))
+        regexTokenized.select(F.size(F.col("token_list")).alias("word_count"))
         .agg(F.max(F.col("word_count")).alias("max_review_length"))
         .collect()
     )

examples/reviews/implementations/aggregators/vocab.py

@@ -1,15 +1,12 @@
 def aggregate_spark(data, columns, args):
     import pyspark.sql.functions as F
-    from pyspark.ml.feature import StopWordsRemover, RegexTokenizer
+    from pyspark.ml.feature import RegexTokenizer
 
-    input_data = data.withColumn(columns["col"], F.lower(F.col(columns["col"])))
     regexTokenizer = RegexTokenizer(inputCol=columns["col"], outputCol="token_list", pattern="\\W")
     regexTokenized = regexTokenizer.transform(data)
 
-    remover = StopWordsRemover(inputCol="token_list", outputCol="filtered_word_list")
     vocab_rows = (
-        remover.transform(regexTokenized)
-        .select(F.explode(F.col("filtered_word_list")).alias("word"))
+        regexTokenized.select(F.explode(F.col("token_list")).alias("word"))
         .groupBy("word")
         .count()
         .orderBy(F.col("count").desc())
@@ -19,6 +16,7 @@ def aggregate_spark(data, columns, args):
     )
 
     vocab = [row["word"] for row in vocab_rows]
-    reverse_dict = {word: idx + len(args["reserved_indices"]) for idx, word in enumerate(vocab)}
-
-    return {**reverse_dict, **args["reserved_indices"]}
+    reverse_dict = {word: 2 + idx for idx, word in enumerate(vocab)}
+    reverse_dict["<PAD>"] = 0
+    reverse_dict["<UNKNOWN>"] = 1
+    return reverse_dict

examples/reviews/implementations/models/t2t_transformer.py

@@ -0,0 +1,65 @@
+import tensorflow as tf
+from tensor2tensor.utils import trainer_lib
+from tensor2tensor import models  # pylint: disable=unused-import
+from tensor2tensor import problems  # pylint: disable=unused-import
+from tensor2tensor.data_generators import problem_hparams
+from tensor2tensor.utils import registry
+from tensor2tensor.utils import metrics
+from tensor2tensor.data_generators import imdb
+from tensor2tensor.data_generators import text_encoder
+
+
+def create_estimator(run_config, model_config):
+    # t2t expects these keys in run_config
+    run_config.data_parallelism = None
+    run_config.t2t_device_info = {"num_async_replicas": 1}
+
+    hparams = trainer_lib.create_hparams("transformer_base_single_gpu")
+
+    problem = SentimentIMDBCortex(list(model_config["aggregates"]["reviews_vocab"]))
+    p_hparams = problem.get_hparams(hparams)
+    hparams.problem = problem
+    hparams.problem_hparams = p_hparams
+
+    problem.eval_metrics = lambda: [
+        metrics.Metrics.ACC_TOP5,
+        metrics.Metrics.ACC_PER_SEQ,
+        metrics.Metrics.NEG_LOG_PERPLEXITY,
+    ]
+
+    # t2t expects this key
+    hparams.warm_start_from = None
+
+    # reduce memory load
+    hparams.num_hidden_layers = 2
+    hparams.hidden_size = 32
+    hparams.filter_size = 32
+    hparams.num_heads = 2
+
+    estimator = trainer_lib.create_estimator("transformer", hparams, run_config)
+    return estimator
+
+
+def transform_tensorflow(features, labels, model_config):
+    max_length = model_config["aggregates"]["max_review_length"]
+
+    features["inputs"] = tf.expand_dims(tf.reshape(features["embedding_input"], [max_length]), -1)
+    features["targets"] = tf.expand_dims(tf.expand_dims(labels, -1), -1)
+
+    return features, labels
+
+
+class SentimentIMDBCortex(imdb.SentimentIMDB):
+    """IMDB sentiment classification, with an in-memory vocab"""
+
+    def __init__(self, vocab_list):
+        super().__init__()
+        self.vocab = vocab_list
+
+    def feature_encoders(self, data_dir):
+        encoder = text_encoder.TokenTextEncoder(vocab_filename=None, vocab_list=self.vocab)
+
+        return {
+            "inputs": encoder,
+            "targets": text_encoder.ClassLabelEncoder(self.class_labels(data_dir)),
+        }

examples/reviews/implementations/transformers/tokenize_string_to_int.py

@@ -6,21 +6,16 @@
 def transform_python(sample, args):
     text = sample["col"].lower()
     token_index_list = []
-
-    reverse_vocab = args["vocab"]
-    stop_words = args["stop_words"]
-    reserved_indices = args["reserved_indices"]
+    vocab = args["vocab"]
 
     for token in non_word.split(text):
         if len(token) == 0:
             continue
-        if token in stop_words:
-            continue
-        token_index_list.append(reverse_vocab.get(token, reserved_indices["<UNKNOWN>"]))
+        token_index_list.append(vocab.get(token, vocab["<UNKNOWN>"]))
         if len(token_index_list) == args["max_len"]:
             break
 
     for i in range(args["max_len"] - len(token_index_list)):
-        token_index_list.append(reserved_indices["<PAD>"])
+        token_index_list.append(vocab["<PAD>"])
 
     return token_index_list

examples/reviews/requirements.txt

@@ -0,0 +1 @@
+tensor2tensor==1.10.0