<jupyter_start><jupyter_text>Inside Account A Read/Write to its own Online and Offline Store  Prerequisites<jupyter_code>#!pip install awswrangler<jupyter_output><empty_output><jupyter_text>Imports<jupyter_code>from sagemaker.feature_store.feature_group import FeatureGroup
from sagemaker import get_execution_role
from sagemaker.session import Session
import awswrangler as wr
import pandas as pd
import sagemaker
import logging
import boto3
import time
import s3fs<jupyter_output><empty_output><jupyter_text>Setup Logger<jupyter_code>logger = logging.getLogger('sagemaker')
logger.setLevel(logging.INFO)
logger.addHandler(logging.StreamHandler())
logger.info(f'[Using SageMaker version: {sagemaker.__version__}]')<jupyter_output>[Using SageMaker version: 2.16.4.dev0]<jupyter_text>Essentials * Create SageMaker & Feature Store Runtime Clients* Create a Feature Store Session encapsulating the above clients* Ensure the Execution Role you use for this notebook has all the required policies attached to it as per the instructions. If not, please make sure to attach them to the role before proceeding.<jupyter_code>region = boto3.Session().region_name
boto_session = boto3.Session(region_name=region)
s3 = boto_session.resource('s3', region_name=region)
role = get_execution_role()

s3_client = boto3.client('s3', region_name=region)
sagemaker_client = boto_session.client(service_name='sagemaker', region_name=region)
featurestore_runtime = boto_session.client(service_name='sagemaker-featurestore-runtime', region_name=region)<jupyter_output><empty_output><jupyter_text>https://sagemaker.readthedocs.io/en/stable/amazon_sagemaker_featurestore.html API Documentation: https://sagemaker.readthedocs.io/en/stable/api/prep_data/feature_store.html<jupyter_code>feature_store_session = Session(
    boto_session=boto_session,
    sagemaker_client=sagemaker_client,
    sagemaker_featurestore_runtime_client=featurestore_runtime
)<jupyter_output><empty_output><jupyter_text>Pick a Bucket to be used as Offline Feature StoreIMPORTANT: If the bucket name does not contain the word `sagemaker`, make sure to change the bucket name inside `AmazonSageMakerFeatureStoreAccess` policy to capture it.<jupyter_code>bucket = 'sagemaker-feature-store-account-a'
if s3.Bucket(bucket).creation_date is None:
    s3.create_bucket(Bucket=bucket)
else:
    logger.warn(f'Bucket: {bucket} already exists!')<jupyter_output>Bucket: sagemaker-feature-store-account-a already exists!<jupyter_text>`offline_feature_store_s3_uri` URI below is the location of your offline store<jupyter_code>offline_feature_store_s3_uri = f's3://{bucket}/'
offline_feature_store_s3_uri<jupyter_output><empty_output><jupyter_text>Load Features<jupyter_code>features = pd.read_csv('features.csv', names=['employee_id', 'name', 'age', 'sex', 'happiness_score'])
features['created_by'] = 'account-a'
features.dtypes<jupyter_output><empty_output><jupyter_text>Ingest Features into SageMaker Feature Store<jupyter_code>record_identifier_feature_name = 'employee_id'
event_time_feature_name = 'event_time'<jupyter_output><empty_output><jupyter_text>Create Feature Group<jupyter_code>feature_group_name = 'employees'
feature_group = FeatureGroup(name=feature_group_name, sagemaker_session=feature_store_session)<jupyter_output><empty_output><jupyter_text>Feature Store supported types are `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>def cast_object_to_string(df):
    """
    Cast object dtype to string. The SageMaker FeatureStore Python SDK will then 
    map the string dtype to String feature type.
    """
    for label in df.columns:
        if df.dtypes[label] == 'object':
            df[label] = df[label].astype('string')
cast_object_to_string(features)<jupyter_output><empty_output><jupyter_text>Append event_time to the `features` dataframe<jupyter_code>current_time_sec = int(round(time.time()))
features[event_time_feature_name] = pd.Series([current_time_sec]*len(features), dtype='float64')
features.dtypes
features<jupyter_output><empty_output><jupyter_text>Load Feature DefinitionsSageMaker FeatureStore Python SDK will auto-detect the data schema based on input data<jupyter_code>feature_group.load_feature_definitions(data_frame=features)<jupyter_output><empty_output><jupyter_text>KMS Custom Key ID Use the KMS Customer Managed Key (CMK) for encrypting the data inside the feature store.<jupyter_code>kms_key = 'arn:aws:kms:us-east-1:892313895307:key/d3763b61-8d94-43bd-a3d6-4b4516ad28e7'<jupyter_output><empty_output><jupyter_text>Create Feature Group Delete feature group if it already exists by uncommenting the line below.<jupyter_code>#sagemaker_client.delete_feature_group(FeatureGroupName='employees')<jupyter_output><empty_output><jupyter_text>Create the feature group named `employees`<jupyter_code>feature_group.create(
    s3_uri=offline_feature_store_s3_uri,
    record_identifier_name=record_identifier_feature_name,
    event_time_feature_name=event_time_feature_name,
    role_arn=role,
    enable_online_store=True,
    offline_store_kms_key_id=kms_key
)<jupyter_output><empty_output><jupyter_text>Note: Ensure `FeatureGroupStatus` of the created feature group shows as `Created` before proceeding to the next steps.  Validate if feature group is created<jupyter_code>feature_group.describe()
sagemaker_client.list_feature_groups()<jupyter_output><empty_output><jupyter_text>Put Records into Feature Group (Both Online & Offline) After 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. For this example, to accelerate the ingestion process, we are specifying multiple workers to do the job simultaneously.<jupyter_code>%%time

feature_group.ingest(data_frame=features, max_workers=5, wait=True)<jupyter_output>Started ingesting index 0 to 1
Started ingesting index 1 to 2
Started ingesting index 2 to 3
Started ingesting index 3 to 4
Started ingesting index 4 to 5
Successfully ingested row 0 to 1
Successfully ingested row 4 to 5
Successfully ingested row 2 to 3
Successfully ingested row 1 to 2
Successfully ingested row 3 to 4<jupyter_text>Get Record from Online Store (Available Immediately) To confirm that data has been ingested, we can quickly retrieve a record from the online store:<jupyter_code>record_identifier = str(101)

featurestore_runtime.get_record(FeatureGroupName=feature_group_name, 
                                RecordIdentifierValueAsString=record_identifier)<jupyter_output><empty_output><jupyter_text>Get Records from Offline StoreNow let's wait for the data to appear in our offline store before moving forward to creating a dataset. This will take approximately 5 minutes.<jupyter_code>account_id = boto3.client('sts').get_caller_identity()['Account']
feature_group_s3_prefix = f'{account_id}/sagemaker/{region}/offline-store/{feature_group_name}/data'
feature_group_s3_prefix
offline_store_contents = None
while offline_store_contents is None:
    objects = s3_client.list_objects(Bucket=bucket, Prefix=feature_group_s3_prefix)
    if 'Contents' in objects and len(objects['Contents']) > 1:
        logger.info('[Features are available in Offline Store!]')
        offline_store_contents = objects['Contents']
    else:
        logger.info('[Waiting for data in Offline Store...]')
        time.sleep(60)
offline_store_contents<jupyter_output><empty_output><jupyter_text>Inspect the Parquet Files (Offline Store) using AWS Wrangler<jupyter_code>s3_prefix = '/'.join(offline_store_contents[0]['Key'].split('/')[:-5])
s3_uri = f's3://{bucket}/{s3_prefix}'
s3_uri
df = wr.s3.read_parquet(path=s3_uri)
df<jupyter_output><empty_output><jupyter_text>Build Train Set from Offline Store Features using Glue and AthenaSageMaker FeatureStore automatically builds the `Glue Data Catalog` for FeatureGroups (you can optionally turn it on/off while creating the FeatureGroup). In this example, we want to create one training dataset with FeatureValues from the feature group we created. This is done by utilizing the auto-built Catalog. We can then run an `Athena` query that pulls the data stored in the offline store in S3 from the FeatureGroup.<jupyter_code>query = feature_group.athena_query()
query.__dict__
table = query.table_name
table
query_string = f'SELECT * FROM "{table}"'
print('Running ' + query_string)
query.run(query_string=query_string, output_location=f's3://{bucket}/{account_id}/query_results/')
query.wait()
df = query.as_dataframe()[['employee_id', 'name', 'age', 'sex', 'happiness_score', 'created_by']]
df<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.08573027420257415, 'fraction_numerical': 0.01152770005595971, 'mean_word_length': 7.274074074074074, 'pattern_counts': {'":': 0, '<': 86, '<?xml version=': 0, '>': 87, 'https://': 2, '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': '22', 'dir': 'jupyter-structured-clean-dedup', 'id': '190189', 'license': 'Apache-2.0', 'n_tokens_mistral': 2881, 'n_tokens_neox': 2763, 'n_words': 728, 'path': 'pattern-2/account-a/self-access.ipynb', 'repo_name': 'arunprsh/sagemaker-feature-store-cross-account-access', 'size': '8935'}