<jupyter_start><jupyter_text>Data Wrangler Feature Store Notebook Use this notebook to create a feature group and add features to an offline or onlinefeature store using a Data Wrangler .flow file.A single *feature* corresponds to a column in your dataset. A *feature group* is a predefinedschema for a collection of features - each feature in the feature group has a specified datatype and name. A single *record* in a feature group corresponds to a row in your datataframe.A *feature store* is a collection of feature groups.This notebook uses Amazon SageMaker Feature Store (Feature Store) to create a feature groupand ingest data into feature store. To learn more about SageMaker Feature Store, see[Amazon Feature Store Documentation](http://docs.aws.amazon.com/sagemaker/latest/dg/feature-store.html).To create a feature group, you will create the following resources:* A feature definition using a schema, record identifier, and event-time feature name.* An online or offline store configuration.You will use a processing job to process your data at scale and ingest the data into this feature group. First, use the following cell to install dependencies.<jupyter_code># SageMaker Python SDK version 2.x is required
import pkg_resources
import subprocess
import sys

original_version = pkg_resources.get_distribution("sagemaker").version
_ = subprocess.check_call(
    [sys.executable, "-m", "pip", "install", "sagemaker==2.20.0"]
    )
import os
import uuid
import json
import time
import boto3
import sagemaker<jupyter_output><empty_output><jupyter_text>ParametersThe following lists configurable parameters that are used throughout this notebook.<jupyter_code># S3 bucket for saving processing job outputs
# Feel free to specify a different bucket here if you wish.
sess = sagemaker.Session()
bucket = sess.default_bucket()
prefix = "data_wrangler_flows"
flow_id = f"{time.strftime('%d-%H-%M-%S', time.gmtime())}-{str(uuid.uuid4())[:8]}"
flow_name = f"flow-sm-tutorial-{flow_id}"  # CHANGED ********
flow_uri = f"s3://{bucket}/{prefix}/{flow_name}.flow"

flow_file_name = "./01_data_prep/wrangler.flow" # CHANGED ********

iam_role = sagemaker.get_execution_role()

container_uri = "663277389841.dkr.ecr.us-east-1.amazonaws.com/sagemaker-data-wrangler-container:1.x"

# Processing Job Resources Configurations
processing_job_name = f"data-wrangler-feature-store-processing-{flow_id}"
processing_dir = "/opt/ml/processing"

# URL to use for sagemaker client.
# If this is None, boto will automatically construct the appropriate URL to use
# when communicating with sagemaker.
sagemaker_endpoint_url = None
iam_role<jupyter_output><empty_output><jupyter_text>Push Flow to S3Use the following cell to upload the Data Wrangler .flow file to Amazon S3 so thatit can be used as an input to the processing job.<jupyter_code># Load .flow file
with open(flow_file_name) as f:
    flow = json.load(f)

# Upload to S3
s3_client = boto3.client("s3")
s3_client.upload_file(flow_file_name, bucket, f"{prefix}/{flow_name}.flow")

print(f"Data Wrangler Flow notebook uploaded to {flow_uri}")<jupyter_output>Data Wrangler Flow notebook uploaded to s3://sagemaker-us-east-1-367158743199/data_wrangler_flows/flow-sm-tutorial-31-16-16-17-9f41d66b.flow<jupyter_text>Create Feature Group<jupyter_code>feature_group_name = f'FG-{flow_name}'
print(f"Feature Group Name: {feature_group_name}")

%store feature_group_name<jupyter_output>Feature Group Name: FG-flow-sm-tutorial-31-16-16-17-9f41d66b<jupyter_text>The following cell maps types between Data Wrangler supported types and Feature Storesupported types (`String`, `Fractional`, and `Integral`). The default type is set to `String`.This means that, if a column in your dataset is not a `float` or `long` type,it will default to `String` in your Feature Store.<jupyter_code>datawrangler_FG_type_mapping = {
    'float': 'Fractional',
    'long': 'Integral'
}

# Some schema types in Data Wrangler are not supported by Feature Store.
# Feature store supports String, Integral, and Fractional types.
# The following will create a default_FG_type set to String for these types.
default_FG_type = "String"<jupyter_output><empty_output><jupyter_text>The following is a list of the column names and data types of the final dataset that will be producedwhen your data flow is used to process your input dataset.<jupyter_code>column_schema = [
    {
        "name": "LABEL",
        "type": "float"
    },
    {
        "name": "ID",
        "type": "float"
    },
    {
        "name": "LIMIT_BAL",
        "type": "float"
    },
    {
        "name": "SEX",
        "type": "float"
    },
    {
        "name": "EDUCATION",
        "type": "float"
    },
    {
        "name": "MARRIAGE",
        "type": "float"
    },
    {
        "name": "AGE",
        "type": "float"
    },
    {
        "name": "PAY_0",
        "type": "float"
    },
    {
        "name": "PAY_2",
        "type": "float"
    },
    {
        "name": "PAY_3",
        "type": "float"
    },
    {
        "name": "PAY_4",
        "type": "float"
    },
    {
        "name": "PAY_5",
        "type": "float"
    },
    {
        "name": "PAY_6",
        "type": "float"
    },
    {
        "name": "BILL_AMT1",
        "type": "float"
    },
    {
        "name": "BILL_AMT2",
        "type": "float"
    },
    {
        "name": "BILL_AMT3",
        "type": "float"
    },
    {
        "name": "BILL_AMT4",
        "type": "float"
    },
    {
        "name": "BILL_AMT5",
        "type": "float"
    },
    {
        "name": "BILL_AMT6",
        "type": "float"
    },
    {
        "name": "PAY_AMT1",
        "type": "float"
    },
    {
        "name": "PAY_AMT2",
        "type": "float"
    },
    {
        "name": "PAY_AMT3",
        "type": "float"
    },
    {
        "name": "PAY_AMT4",
        "type": "float"
    },
    {
        "name": "PAY_AMT5",
        "type": "float"
    },
    {
        "name": "PAY_AMT6",
        "type": "float"
    },
    {
        "name": "EVENT_TIME",
        "type": "float"
    }
]<jupyter_output><empty_output><jupyter_text>Select Record identifier and Event time feature name. These are required parameters for feature groupcreation.* **Record identifier name** is the name of the feature whose value uniquely identiﬁes a Recorddeﬁned in the feature group's feature definitions.* **Event time feature name** is the name of the EventTime of a Record in FeatureGroup.A EventTime is point in time when a new event occurs that corresponds to the creation or update of aRecord in FeatureGroup. All Records in the FeatureGroup must have a corresponding EventTime.<jupyter_code>record_identifier_name = 'ID' # CHANGED ********
if record_identifier_name is None:
   raise RuntimeError("Select a column name as the feature group identifier.")

event_time_feature_name = 'EVENT_TIME' # CHANGED ********
if event_time_feature_name is None:
   raise RuntimeError("Select a column name as the event time feature name.")

# Below you map the schema detected from Data Wrangler to Feature Group Types.
feature_definitions = [
    {
        "FeatureName": schema['name'],
        "FeatureType": datawrangler_FG_type_mapping.get(
            schema['type'],
            default_FG_type
         )
    } for schema in column_schema
]
print(feature_definitions)<jupyter_output>[{'FeatureName': 'LABEL', 'FeatureType': 'Fractional'}, {'FeatureName': 'ID', 'FeatureType': 'Fractional'}, {'FeatureName': 'LIMIT_BAL', 'FeatureType': 'Fractional'}, {'FeatureName': 'SEX', 'FeatureType': 'Fractional'}, {'FeatureName': 'EDUCATION', 'FeatureType': 'Fractional'}, {'FeatureName': 'MARRIAGE', 'FeatureType': 'Fractional'}, {'FeatureName': 'AGE', 'FeatureType': 'Fractional'}, {'FeatureName': 'PAY_0', 'FeatureType': 'Fractional'}, {'FeatureName': 'PAY_2', 'FeatureType': 'Fractional'}, {'FeatureName': 'PAY_3', 'FeatureType': 'Fractional'}, {'FeatureName': 'PAY_4', 'FeatureType': 'Fractional'}, {'FeatureName': 'PAY_5', 'FeatureType': 'Fractional'}, {'FeatureName': 'PAY_6', 'FeatureType': 'Fractional'}, {'FeatureName': 'BILL_AMT1', 'FeatureType': 'Fractional'}, {'FeatureName': 'BILL_AMT2', 'FeatureType': 'Fractional'}, {'FeatureName': 'BILL_AMT3', 'FeatureType': 'Fractional'}, {'FeatureName': 'BILL_AMT4', 'FeatureType': 'Fractional'}, {'FeatureName': 'BILL_AMT5', 'FeatureType': [...]<jupyter_text>The following are your online and offline store configurations. You enable an onlinestore by setting `EnableOnlineStore` to `True`. The offline store is located in anAmazon S3 bucket in your account. To update the bucket used, update theparameter `bucket` in the second code cell in this notebook.<jupyter_code>sagemaker_client = boto3.client("sagemaker", endpoint_url=sagemaker_endpoint_url)

# Online Store Configuration
online_store_config = {
    "EnableOnlineStore": True
}

# Offline Store Configuration
s3_uri = 's3://' + bucket # this is the default bucket defined in previous cells
offline_store_config = {
    "S3StorageConfig": {
        "S3Uri": s3_uri
    }
}

# Create Feature Group
create_fg_response = sagemaker_client.create_feature_group(
    FeatureGroupName = feature_group_name,
    EventTimeFeatureName = event_time_feature_name,
    RecordIdentifierFeatureName = record_identifier_name,
    FeatureDefinitions = feature_definitions,
    OnlineStoreConfig = online_store_config,
    OfflineStoreConfig = offline_store_config,
    RoleArn = iam_role)

# Describe Feature Group
status = sagemaker_client.describe_feature_group(FeatureGroupName=feature_group_name)
while status['FeatureGroupStatus'] != 'Created':
    if status['FeatureGroupStatus'] == 'CreateFailed':
        raise RuntimeError(f"Feature Group Creation Failed: {status}")
    status = sagemaker_client.describe_feature_group(FeatureGroupName=feature_group_name)
    print("Feature Group Status: " + status['FeatureGroupStatus'])
    time.sleep(3)

print(status)<jupyter_output>Feature Group Status: Creating
Feature Group Status: Creating
Feature Group Status: Creating
Feature Group Status: Creating
Feature Group Status: Created
{'FeatureGroupArn': 'arn:aws:sagemaker:us-east-1:367158743199:feature-group/fg-flow-sm-tutorial-31-16-16-17-9f41d66b', 'FeatureGroupName': 'FG-flow-sm-tutorial-31-16-16-17-9f41d66b', 'RecordIdentifierFeatureName': 'ID', 'EventTimeFeatureName': 'EVENT_TIME', 'FeatureDefinitions': [{'FeatureName': 'LABEL', 'FeatureType': 'Fractional'}, {'FeatureName': 'ID', 'FeatureType': 'Fractional'}, {'FeatureName': 'LIMIT_BAL', 'FeatureType': 'Fractional'}, {'FeatureName': 'SEX', 'FeatureType': 'Fractional'}, {'FeatureName': 'EDUCATION', 'FeatureType': 'Fractional'}, {'FeatureName': 'MARRIAGE', 'FeatureType': 'Fractional'}, {'FeatureName': 'AGE', 'FeatureType': 'Fractional'}, {'FeatureName': 'PAY_0', 'FeatureType': 'Fractional'}, {'FeatureName': 'PAY_2', 'FeatureType': 'Fractional'}, {'FeatureName': 'PAY_3', 'FeatureType': 'Fractional'}, {'FeatureNa[...]<jupyter_text>Use the following code cell to define helper functions for creating inputs toa processing job.<jupyter_code>def create_flow_notebook_processing_input(base_dir, flow_s3_uri):
    return {
        "InputName": "flow",
        "S3Input": {
            "LocalPath": f"{base_dir}/flow",
            "S3Uri": flow_s3_uri,
            "S3DataType": "S3Prefix",
            "S3InputMode": "File",
        },
    }

def create_s3_processing_input(base_dir, name, dataset_definition):
    return {
        "InputName": name,
        "S3Input": {
            "LocalPath": f"{base_dir}/{name}",
            "S3Uri": dataset_definition["s3ExecutionContext"]["s3Uri"],
            "S3DataType": "S3Prefix",
            "S3InputMode": "File",
        },
    }

def create_redshift_processing_input(base_dir, name, dataset_definition):
    return {
        "InputName": name,
        "DatasetDefinition": {
            "RedshiftDatasetDefinition": {
                "ClusterId": dataset_definition["clusterIdentifier"],
                "Database": dataset_definition["database"],
                "DbUser": dataset_definition["dbUser"],
                "QueryString": dataset_definition["queryString"],
                "ClusterRoleArn": dataset_definition["unloadIamRole"],
                "OutputS3Uri": f'{dataset_definition["s3OutputLocation"]}{name}/',
                "OutputFormat": dataset_definition["outputFormat"].upper(),
            },
            "LocalPath": f"{base_dir}/{name}",
        },
    }

def create_athena_processing_input(base_dir, name, dataset_definition):
    return {
        "InputName": name,
        "DatasetDefinition": {
            "AthenaDatasetDefinition": {
                "Catalog": dataset_definition["catalogName"],
                "Database": dataset_definition["databaseName"],
                "QueryString": dataset_definition["queryString"],
                "OutputS3Uri": f'{dataset_definition["s3OutputLocation"]}{name}/',
                "OutputFormat": dataset_definition["outputFormat"].upper(),
            },
            "LocalPath": f"{base_dir}/{name}",
        },
    }

def create_processing_inputs(processing_dir, flow, flow_uri):
    """Helper function for creating processing inputs
    :param flow: loaded data wrangler flow notebook
    :param flow_uri: S3 URI of the data wrangler flow notebook
    """
    processing_inputs = []
    flow_processing_input = create_flow_notebook_processing_input(processing_dir, flow_uri)
    processing_inputs.append(flow_processing_input)

    for node in flow["nodes"]:
        if "dataset_definition" in node["parameters"]:
            data_def = node["parameters"]["dataset_definition"]
            name = data_def["name"]
            source_type = data_def["datasetSourceType"]

            if source_type == "S3":
                s3_processing_input = create_s3_processing_input(
                    processing_dir, name, data_def)
                processing_inputs.append(s3_processing_input)
            elif source_type == "Athena":
                athena_processing_input = create_athena_processing_input(
                    processing_dir, name, data_def)
                processing_inputs.append(athena_processing_input)
            elif source_type == "Redshift":
                redshift_processing_input = create_redshift_processing_input(
                    processing_dir, name, data_def)
                processing_inputs.append(redshift_processing_input)
            else:
                raise ValueError(f"{source_type} is not supported for Data Wrangler Processing.")
    return processing_inputs<jupyter_output><empty_output><jupyter_text>Start ProcessingJobNow, the Processing Job is submitted to a boto client. The status of the processing job ismonitored with the boto client, and this notebook waits until the job is no longer 'InProgress'.<jupyter_code># Processing job name
print(f'Processing Job Name: {processing_job_name}')

processingResources = {
        'ClusterConfig': {
            'InstanceCount': 1,
            'InstanceType': 'ml.m5.4xlarge',
            'VolumeSizeInGB': 30
        }
    }

appSpecification = {'ImageUri': container_uri}

sagemaker_client.create_processing_job(
        ProcessingInputs=create_processing_inputs(processing_dir, flow, flow_uri),
        ProcessingOutputConfig={
            'Outputs': [
                {
                    'OutputName': 'a23a9cd7-bf8c-40b0-be11-2adfa82f0632.default',
                    'FeatureStoreOutput': {
                        'FeatureGroupName': feature_group_name
                    },
                    'AppManaged': True
                }
            ],
        },
        ProcessingJobName=processing_job_name,
        ProcessingResources=processingResources,
        AppSpecification=appSpecification,
        RoleArn=iam_role
    )


status = sagemaker_client.describe_processing_job(ProcessingJobName=processing_job_name)

while status['ProcessingJobStatus'] in ('InProgress', 'Failed'):
    if status['ProcessingJobStatus'] == 'Failed':
        raise RuntimeError(f"Processing Job failed: {status}")
    status = sagemaker_client.describe_processing_job(ProcessingJobName=processing_job_name)
    print(status['ProcessingJobStatus'])
    time.sleep(60)

print(status)<jupyter_output>Processing Job Name: data-wrangler-feature-store-processing-31-16-16-17-9f41d66b
InProgress
InProgress
InProgress
InProgress
InProgress
InProgress
Completed
{'ProcessingInputs': [{'InputName': 'flow', 'AppManaged': False, 'S3Input': {'S3Uri': 's3://sagemaker-us-east-1-367158743199/data_wrangler_flows/flow-sm-tutorial-31-16-16-17-9f41d66b.flow', 'LocalPath': '/opt/ml/processing/flow', 'S3DataType': 'S3Prefix', 'S3InputMode': 'File', 'S3DataDistributionType': 'FullyReplicated'}}, {'InputName': 'dataset.csv', 'AppManaged': False, 'S3Input': {'S3Uri': 's3://sagemaker-us-east-1-367158743199/sagemaker-tutorial/data/dataset.csv', 'LocalPath': '/opt/ml/processing/dataset.csv', 'S3DataType': 'S3Prefix', 'S3InputMode': 'File', 'S3DataDistributionType': 'FullyReplicated'}}], 'ProcessingOutputConfig': {'Outputs': [{'OutputName': 'a23a9cd7-bf8c-40b0-be11-2adfa82f0632.default', 'FeatureStoreOutput': {'FeatureGroupName': 'FG-flow-sm-tutorial-31-16-16-17-9f41d66b'}, 'AppManaged': True}]}, 'ProcessingJ[...]<jupyter_text>CleanupUncomment the following code cell to revert the SageMaker Python SDK to the original version usedbefore running this notebook. This notebook upgrades the SageMaker Python SDK to 2.x, which maycause other example notebooks to break. To learn more about the changes introduced in theSageMaker Python SDK 2.x update, see[Use Version 2.x of the SageMaker Python SDK.](https://sagemaker.readthedocs.io/en/stable/v2.html).<jupyter_code># _ = subprocess.check_call(
#         [sys.executable, "-m", "pip", "install", f"sagemaker=={original_version}"]
#     )<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.1248736664795059, 'fraction_numerical': 0.016844469399213923, 'mean_word_length': 3.7969297064368437, 'pattern_counts': {'":': 92, '<': 40, '<?xml version=': 0, '>': 40, 'https://': 1, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '11', 'dir': 'jupyter-structured-clean-dedup', 'id': '261673', 'license': 'Apache-2.0', 'n_tokens_mistral': 5582, 'n_tokens_neox': 5375, 'n_words': 1342, 'path': 'notebooks/01_data_prep/wrangler_notebook/013_wrangler_to_featurestore.ipynb', 'repo_name': 'shashankprasanna/sagemaker-workshop-2021', 'size': '17810'}