<jupyter_start><jupyter_text>Feature Transformation with Scikit-Learn In This Notebook Saving Features into the SageMaker Feature StoreIn this notebook, we convert raw text into BERT embeddings.  This will allow us to perform natural language processing tasks such as text classification. We save the features into the SageMaker Feature Store. ![](img/prepare_dataset_bert.png)<jupyter_code>import sagemaker
import boto3

sess = sagemaker.Session()
bucket = sess.default_bucket()
role = sagemaker.get_execution_role()
region = boto3.Session().region_name

sm = boto3.Session().client(service_name="sagemaker", region_name=region)
s3 = boto3.Session().client(service_name="s3", region_name=region)<jupyter_output><empty_output><jupyter_text>![BERT Mania](img/bert_mania.png)  Convert Raw Text to BERT Features using Hugging Face and TensorFlow<jupyter_code>import tensorflow as tf
import collections
import json
import os
import pandas as pd
import csv
from transformers import DistilBertTokenizer

tokenizer = DistilBertTokenizer.from_pretrained("distilbert-base-uncased")

REVIEW_BODY_COLUMN = "review_body"
REVIEW_ID_COLUMN = "review_id"
# DATE_COLUMN = 'date'

LABEL_COLUMN = "star_rating"
LABEL_VALUES = [1, 2, 3, 4, 5]

label_map = {}
for (i, label) in enumerate(LABEL_VALUES):
    label_map[label] = i


class InputFeatures(object):
    """BERT feature vectors."""

    def __init__(self, input_ids, input_mask, segment_ids, label_id, review_id, date, label):
        self.input_ids = input_ids
        self.input_mask = input_mask
        self.segment_ids = segment_ids
        self.label_id = label_id
        self.review_id = review_id
        self.date = date
        self.label = label
        

class Input(object):
    """A single training/test input for sequence classification."""

    def __init__(self, text, review_id, date, label=None):
        """Constructs an Input.
        Args:
          text: string. The untokenized text of the first sequence. For single
            sequence tasks, only this sequence must be specified.
          label: (Optional) string. The label of the example. This should be
            specified for train and dev examples, but not for test examples.
        """
        self.text = text
        self.review_id = review_id
        self.date = date
        self.label = label


def convert_input(the_input, max_seq_length):
    # First, we need to preprocess our data so that it matches the data BERT was trained on:
    # 1. Lowercase our text (if we're using a BERT lowercase model)
    # 2. Tokenize it (i.e. "sally says hi" -> ["sally", "says", "hi"])
    # 3. Break words into WordPieces (i.e. "calling" -> ["call", "##ing"])
    #
    # Fortunately, the Transformers tokenizer does this for us!

    tokens = tokenizer.tokenize(the_input.text)
    tokens.insert(0, '[CLS]')
    tokens.append('[SEP]')
    print("**{} tokens**\n{}\n".format(len(tokens), tokens))

    encode_plus_tokens = tokenizer.encode_plus(
        the_input.text,
        pad_to_max_length=True,
        max_length=max_seq_length,
        truncation=True
    )

    # The id from the pre-trained BERT vocabulary that represents the token.  (Padding of 0 will be used if the # of tokens is less than `max_seq_length`)
    input_ids = encode_plus_tokens["input_ids"]

    # Specifies which tokens BERT should pay attention to (0 or 1).  Padded `input_ids` will have 0 in each of these vector elements.
    input_mask = encode_plus_tokens["attention_mask"]

    # Segment ids are always 0 for single-sequence tasks such as text classification.  1 is used for two-sequence tasks such as question/answer and next sentence prediction.
    segment_ids = [0] * max_seq_length

    # Label for each training row (`star_rating` 1 through 5)
    label_id = label_map[the_input.label]

    features = InputFeatures(
        input_ids=input_ids,
        input_mask=input_mask,
        segment_ids=segment_ids,
        label_id=label_id,
        review_id=the_input.review_id,
        date=the_input.date,
        label=the_input.label,
    )

    print("**input_ids**\n{}\n".format(features.input_ids))
    print("**input_mask**\n{}\n".format(features.input_mask))
    print("**segment_ids**\n{}\n".format(features.segment_ids))
    print("**label_id**\n{}\n".format(features.label_id))
    print("**review_id**\n{}\n".format(features.review_id))
    print("**date**\n{}\n".format(features.date))
    print("**label**\n{}\n".format(features.label))

    return features


# We'll need to transform our data into a format that BERT understands.
# - `text` is the text we want to classify, which in this case, is the `Request` field in our Dataframe.
# - `label` is the star_rating label (1, 2, 3, 4, 5) for our training input data
def transform_inputs_to_tfrecord(inputs, output_file, max_seq_length):
    records = []
    tf_record_writer = tf.io.TFRecordWriter(output_file)

    for (input_idx, the_input) in enumerate(inputs):
        if input_idx % 10000 == 0:
            print("Writing input {} of {}\n".format(input_idx, len(inputs)))

        features = convert_input(the_input, max_seq_length)

        all_features = collections.OrderedDict()

        # Create TFRecord With input_ids, input_mask, segment_ids, and label_ids
        all_features["input_ids"] = tf.train.Feature(int64_list=tf.train.Int64List(value=features.input_ids))
        all_features["input_mask"] = tf.train.Feature(int64_list=tf.train.Int64List(value=features.input_mask))
        all_features["segment_ids"] = tf.train.Feature(int64_list=tf.train.Int64List(value=features.segment_ids))
        all_features["label_ids"] = tf.train.Feature(int64_list=tf.train.Int64List(value=[features.label_id]))

        tf_record = tf.train.Example(features=tf.train.Features(feature=all_features))
        tf_record_writer.write(tf_record.SerializeToString())

        # Create Record For Feature Store With All Features
        records.append(
            {  #'tf_record': tf_record.SerializeToString(),
                "input_ids": features.input_ids,
                "input_mask": features.input_mask,
                "segment_ids": features.segment_ids,
                "label_id": features.label_id,
                "review_id": the_input.review_id,
                "date": the_input.date,
                "label": features.label,
                #                        'review_body': features.review_body
            }
        )

    tf_record_writer.close()

    return records<jupyter_output><empty_output><jupyter_text>Three(3) feature vectors are created from each raw review (`review_body`) during the feature engineering phase to prepare for BERT processing:* **`input_ids`**:  The id from the pre-trained BERT vocabulary that represents the token.  (Padding of 0 will be used if the  of tokens is less than `max_seq_length`)    * **`input_mask`**:  Specifies which tokens BERT should pay attention to (0 or 1).  Padded `input_ids` will have 0 in each of these vector elements.* **`segment_ids`**:  Segment ids are always 0 for single-sequence tasks such as text classification.  1 is used for two-sequence tasks such as question/answer and next sentence prediction.And one(1) label is created from each raw review (`star_rating`)  :* **`label_id`**:  Label for each training row (`star_rating` 1 through 5)  Demonstrate the BERT-specific Feature Engineering StepWhile we are demonstrating this code with a small amount of data here in the notebook, we will soon scale this to much more data on a powerful SageMaker cluster.  Feature Store requires an Event Time featureWe need a record identifier name and an event time feature name. This will match the column of the corresponding features in our data. Note: Event time date feature type provided Integral. Event time type should be either Fractional(Unix timestamp in seconds) or String (ISO-8601 format) type.<jupyter_code>from datetime import datetime
from time import strftime

# timestamp = datetime.now().replace(microsecond=0).isoformat()
timestamp = datetime.now().strftime("%Y-%m-%dT%H:%M:%SZ")
print(timestamp)
import pandas as pd

data = [
    [
        5,
        "ABCD12345",
        """I needed an "antivirus" application and know the quality of Norton products.  This was a no brainer for me and I am glad it was so simple to get.""",
    ],
    [
        3,
        "EFGH12345",
        """The problem with ElephantDrive is that it requires the use of Java. Since Java is notorious for security problems I haveit removed from all of my computers. What files I do have stored are photos.""",
    ],
    [
        1,
        "IJKL2345",
        """Terrible, none of my codes worked, and I can't uninstall it.  I think this product IS malware and viruses""",
    ],
]

df = pd.DataFrame(data, columns=["star_rating", "review_id", "review_body"])

# Use the InputExample class from BERT's run_classifier code to create examples from the data
inputs = df.apply(
    lambda x: Input(label=x[LABEL_COLUMN], text=x[REVIEW_BODY_COLUMN], review_id=x[REVIEW_ID_COLUMN], date=timestamp),
    axis=1,
)
# Make sure the date is in the correct ISO-8601 format for Feature Store
print(inputs[0].date)<jupyter_output><empty_output><jupyter_text>Save TFRecordsThe three(3) features vectors and one(1) label are converted into a list of `TFRecord` instances (1 per each row of training data):* **`tf_records`**:  Binary representation of each row of training data (3 features + 1 label)These `TFRecord`s are the engineered features that we will use throughout the rest of the pipeline.<jupyter_code>output_file = "./data-tfrecord-featurestore/data.tfrecord"<jupyter_output><empty_output><jupyter_text>Add Features to SageMaker Feature Store  SageMaker Feature Store RuntimeA low-level client representing Amazon SageMaker Feature Store RuntimeContains all data plane API operations and data types for the Amazon SageMaker Feature Store. Use this API to put, delete, and retrieve (get) features from a feature store.<jupyter_code>featurestore_runtime = boto3.Session().client(service_name="sagemaker-featurestore-runtime", region_name=region)<jupyter_output><empty_output><jupyter_text>Create FeatureGroupA feature group is a logical grouping of features, defined in the Feature Store, to describe records. A feature group definition is composed of a list of feature definitions, a record identifier name, and configurations for its online and offline store.Create feature group, describe feature group, update feature groups, delete feature group and list feature groups APIs can be used to manage feature groups.<jupyter_code>from time import gmtime, strftime, sleep

feature_group_name = "reviews-feature-group-" + strftime("%d-%H-%M-%S", gmtime())
print(feature_group_name)
from sagemaker.feature_store.feature_definition import (
    FeatureDefinition,
    FeatureTypeEnum,
)

feature_definitions = [
    FeatureDefinition(feature_name="input_ids", feature_type=FeatureTypeEnum.STRING),
    FeatureDefinition(feature_name="input_mask", feature_type=FeatureTypeEnum.STRING),
    FeatureDefinition(feature_name="segment_ids", feature_type=FeatureTypeEnum.STRING),
    FeatureDefinition(feature_name="label_id", feature_type=FeatureTypeEnum.INTEGRAL),
    FeatureDefinition(feature_name="review_id", feature_type=FeatureTypeEnum.STRING),
    FeatureDefinition(feature_name="date", feature_type=FeatureTypeEnum.STRING),
    FeatureDefinition(feature_name="label", feature_type=FeatureTypeEnum.INTEGRAL),
    #    FeatureDefinition(feature_name='review_body', feature_type=FeatureTypeEnum.STRING),
    FeatureDefinition(feature_name="split_type", feature_type=FeatureTypeEnum.STRING),
]
from sagemaker.feature_store.feature_group import FeatureGroup

feature_group = FeatureGroup(name=feature_group_name, feature_definitions=feature_definitions, sagemaker_session=sess)
print(feature_group)<jupyter_output><empty_output><jupyter_text>Specify `record identifier` and `event time` features<jupyter_code>record_identifier_feature_name = "review_id"
event_time_feature_name = "date"<jupyter_output><empty_output><jupyter_text>Set S3 Prefix for Offline Feature Store<jupyter_code>prefix = "reviews-feature-store-" + timestamp
print(prefix)<jupyter_output><empty_output><jupyter_text>Create Feature GroupThe last step for creating the feature group is to use the `create` function. The online store is not created by default, so we must set this as `True` if we want to enable it. The `s3_uri` is the location of our offline store.<jupyter_code>feature_group.create(
    s3_uri=f"s3://{bucket}/{prefix}",
    record_identifier_name=record_identifier_feature_name,
    event_time_feature_name=event_time_feature_name,
    role_arn=role,
    enable_online_store=False,
)<jupyter_output><empty_output><jupyter_text>Describe the Feature Group<jupyter_code>feature_group.describe()<jupyter_output><empty_output><jupyter_text>List All Feature GroupsWe use the boto3 SageMaker client to list all FeatureGroups.<jupyter_code># sm.list_feature_groups()<jupyter_output><empty_output><jupyter_text>Wait For The Feature Group Creation CompleteCreating a feature group takes time as the data is loaded. We will need to wait until it is created before you can use it. You can check status using the following method.<jupyter_code>import time


def wait_for_feature_group_creation_complete(feature_group):
    status = feature_group.describe().get("FeatureGroupStatus")
    while status == "Creating":
        print("Waiting for Feature Group Creation")
        time.sleep(5)
        status = feature_group.describe().get("FeatureGroupStatus")
    if status != "Created":
        raise RuntimeError(f"Failed to create feature group {feature_group.name}")
    print(f"FeatureGroup {feature_group.name} successfully created.")
wait_for_feature_group_creation_complete(feature_group=feature_group)<jupyter_output><empty_output><jupyter_text>Review The Records To Ingest Into Feature Store<jupyter_code>max_seq_length = 64
records = transform_inputs_to_tfrecord(inputs, output_file, max_seq_length)<jupyter_output><empty_output><jupyter_text>Ingest Records into Feature StoreAfter the FeatureGroups have been created, we can put data into the FeatureGroups by using the `PutRecord` API. This API can handle high TPS and is designed to be called by different streams. The data from all of these Put requests is buffered and written to S3 in chunks. The files will be written to the offline store within a few minutes of ingestion. To accelerate the ingestion process, we can specify multiple workers to do the job simultaneously. Use `put_record(...)` to put a single record in the FeatureGroup.Use `ingest(...)` to ingest the content of a pandas DataFrame to Feature Store. You can set the `max_worker` to the number of threads to be created to work on different partitions of the `data_frame` in parallel.<jupyter_code>import pandas as pd

df_records = pd.DataFrame.from_dict(records)
df_records["split_type"] = "train"
df_records<jupyter_output><empty_output><jupyter_text>Cast DataFrame `Object` to Supported Feature Store Data Type `String`<jupyter_code>def cast_object_to_string(data_frame):
    for label in data_frame.columns:
        if data_frame.dtypes[label] == "object":
            data_frame[label] = data_frame[label].astype("str").astype("string")
cast_object_to_string(df_records)
df_records
feature_group.ingest(data_frame=df_records, max_workers=3, wait=True)<jupyter_output><empty_output><jupyter_text>Wait For Data In Offline Feature Store To Become AvailableCreating a feature group takes time as the data is loaded. We will need to wait until it is created before we can use it.<jupyter_code>offline_store_contents = None

while offline_store_contents is None:
    objects_in_bucket = s3.list_objects(Bucket=bucket, Prefix=prefix)
    if "Contents" in objects_in_bucket and len(objects_in_bucket["Contents"]) > 1:
        offline_store_contents = objects_in_bucket["Contents"]
    else:
        print("Waiting for data in offline store...\n")
        sleep(60)

print("Data available.")<jupyter_output><empty_output><jupyter_text>_Wait For The Cell Above To Complete and show `Data available`._  Get Record From Online Feature StoreUse for OnlineStore serving from a FeatureStore. Only the latest records stored in the OnlineStore can be retrieved. If no Record with `RecordIdentifierValue` is found, then an empty result is returned.<jupyter_code># record_identifier_value = "IJKL2345"

# featurestore_runtime.get_record(
#     FeatureGroupName=feature_group_name, RecordIdentifierValueAsString=record_identifier_value
# )<jupyter_output><empty_output><jupyter_text>Build Training DatasetSageMaker FeatureStore automatically builds the Glue Data Catalog for FeatureGroups (we can optionally turn it on/off while creating the FeatureGroup). We can create a training dataset by querying the data in the feature store. This is done by utilizing the auto-built Catalog and run an Athena query.   Create An Athena Query<jupyter_code>feature_store_query = feature_group.athena_query()<jupyter_output><empty_output><jupyter_text>Get The Feature Group Table Name<jupyter_code>feature_store_table = feature_store_query.table_name<jupyter_output><empty_output><jupyter_text>Build an Athena SQL QueryShow Hive DDL commands to define or change structure of tables or databases in Hive. The schema of the table is generated based on the feature definitions. Columns are named after feature name and data-type are inferred based on feature type. Integral feature type is mapped to INT data-type. Fractional feature type is mapped to FLOAT data-type. String feature type is mapped to STRING data-type.<jupyter_code>print(feature_group.as_hive_ddl())
query_string = """
SELECT input_ids, input_mask, segment_ids, label_id, split_type  FROM "{}" WHERE split_type='train' LIMIT 5
""".format(
    feature_store_table
)

print("Running " + query_string)<jupyter_output><empty_output><jupyter_text>Run Athena QueryThe query results are stored in a S3 bucket.<jupyter_code>feature_store_query.run(query_string=query_string, output_location="s3://" + bucket + "/" + prefix + "/query_results/")

feature_store_query.wait()<jupyter_output><empty_output><jupyter_text>View Query ResultsLoad query results in a Pandas DataFrame.<jupyter_code>dataset = pd.DataFrame()

dataset = feature_store_query.as_dataframe()

dataset<jupyter_output><empty_output><jupyter_text>Review the Feature Store![Feature Store](img/feature_store_sm_extension.png)  Release Resources<jupyter_code>%%html

<p><b>Shutting down your kernel for this notebook to release resources.</b></p>
<button class="sm-command-button" data-commandlinker-command="kernelmenu:shutdown" style="display:none;">Shutdown Kernel</button>
        
<script>
try {
    els = document.getElementsByClassName("sm-command-button");
    els[0].click();
}
catch(err) {
    // NoOp
}    
</script>
%%javascript

try {
    Jupyter.notebook.save_checkpoint();
    Jupyter.notebook.session.delete();
}
catch(err) {
    // NoOp
}<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.09049321695236093, 'fraction_numerical': 0.00615206646335051, 'mean_word_length': 4.761587397758255, 'pattern_counts': {'":': 10, '<': 101, '<?xml version=': 0, '>': 104, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '23', 'dir': 'jupyter-structured-clean-dedup', 'id': '64554', 'license': 'Apache-2.0', 'n_tokens_mistral': 5703, 'n_tokens_neox': 5493, 'n_words': 1713, 'path': '06_prepare/01_Prepare_Dataset_BERT_Scikit_AdHoc_FeatureStore.ipynb', 'repo_name': 'MarcusFra/workshop', 'size': '19018'}