sagemaker.mlops.feature_store.dataset_builder

Dataset Builder for FeatureStore.

Functions

construct_feature_group_to_be_merged(...[, ...])

Construct a FeatureGroupToBeMerged object by provided parameters.

Classes

DatasetBuilder(_sagemaker_session, _base, ...)	DatasetBuilder definition.
FeatureGroupToBeMerged(features, ...[, ...])	FeatureGroup metadata which will be used for SQL join.
JoinComparatorEnum(value)	An enumeration.
JoinTypeEnum(value)	An enumeration.
TableType(value)	An enumeration.

class sagemaker.mlops.feature_store.dataset_builder.DatasetBuilder(_sagemaker_session: Session, _base: FeatureGroup | DataFrame, _output_path: str, _record_identifier_feature_name: str | None = None, _event_time_identifier_feature_name: str | None = None, _included_feature_names: List[str] | None = None, _kms_key_id: str | None = None, _event_time_identifier_feature_type: FeatureTypeEnum | None = None)

Bases: object

DatasetBuilder definition.

This class instantiates a DatasetBuilder object that comprises a base, a list of feature names, an output path and a KMS key ID.

_sagemaker_session

Session instance to perform boto calls.

Type:

Session

_base

A base which can be either a FeatureGroup or a pandas.DataFrame and will be used to merge other FeatureGroups and generate a Dataset.

Type:

Union[FeatureGroup, DataFrame]

_output_path

An S3 URI which stores the output .csv file.

Type:

str

_record_identifier_feature_name

A string representing the record identifier feature if base is a DataFrame (default: None).

Type:

str

_event_time_identifier_feature_name

A string representing the event time identifier feature if base is a DataFrame (default: None).

Type:

str

_included_feature_names

A list of strings representing features to be included in the output. If not set, all features will be included in the output. (default: None).

Type:

List[str]

_kms_key_id

A KMS key id. If set, will be used to encrypt the result file (default: None).

Type:

str

_point_in_time_accurate_join

A boolean representing if point-in-time join is applied to the resulting dataframe when calling “to_dataframe”. When set to True, users can retrieve data using “row-level time travel” according to the event times provided to the DatasetBuilder. This requires that the entity dataframe with event times is submitted as the base in the constructor (default: False).

Type:

bool

_include_duplicated_records

A boolean representing whether the resulting dataframe when calling “to_dataframe” should include duplicated records (default: False).

Type:

bool

_include_deleted_records

A boolean representing whether the resulting dataframe when calling “to_dataframe” should include deleted records (default: False).

Type:

bool

_number_of_recent_records

An integer representing how many records will be returned for each record identifier (default: 1).

Type:

int

_number_of_records

An integer representing the number of records that should be returned in the resulting dataframe when calling “to_dataframe” (default: None).

Type:

int

_write_time_ending_timestamp

A datetime that represents the latest write time for a record to be included in the resulting dataset. Records with a newer write time will be omitted from the resulting dataset. (default: None).

Type:

datetime.datetime

_event_time_starting_timestamp

A datetime that represents the earliest event time for a record to be included in the resulting dataset. Records with an older event time will be omitted from the resulting dataset. (default: None).

Type:

datetime.datetime

_event_time_ending_timestamp

A datetime that represents the latest event time for a record to be included in the resulting dataset. Records with a newer event time will be omitted from the resulting dataset. (default: None).

Type:

datetime.datetime

_feature_groups_to_be_merged

A list of FeatureGroupToBeMerged which will be joined to base (default: []).

Type:

List[FeatureGroupToBeMerged]

_event_time_identifier_feature_type

A FeatureTypeEnum representing the type of event time identifier feature (default: None).

Type:

FeatureTypeEnum

as_of(timestamp: datetime) → DatasetBuilder

Set write_time_ending_timestamp field with provided input.

Parameters:

timestamp (datetime.datetime) – A datetime that all records’ write time in dataset will be before it.

Returns:

This DatasetBuilder object.

classmethod create(base: FeatureGroup | DataFrame, output_path: str, session: Session, record_identifier_feature_name: str | None = None, event_time_identifier_feature_name: str | None = None, included_feature_names: List[str] | None = None, kms_key_id: str | None = None) → DatasetBuilder

Create a DatasetBuilder for generating a Dataset.

Parameters:

• base – A FeatureGroup or DataFrame to use as the base.

• output_path – S3 URI for output.

• session – SageMaker session.

• record_identifier_feature_name – Required if base is DataFrame.

• event_time_identifier_feature_name – Required if base is DataFrame.

• included_feature_names – Features to include in output.

• kms_key_id – KMS key for encryption.

Returns:

DatasetBuilder instance.

include_deleted_records() → DatasetBuilder

Include deleted records in dataset.

Returns:

This DatasetBuilder object.

include_duplicated_records() → DatasetBuilder

Include duplicated records in dataset.

Returns:

This DatasetBuilder object.

point_in_time_accurate_join() → DatasetBuilder

Enable point-in-time accurate join.

Returns:

This DatasetBuilder object.

to_csv_file() → tuple[str, str]

Get query string and result in .csv format file.

Returns:

A tuple containing:

• str: The S3 path of the .csv file

• str: The query string executed

Return type:

tuple

Note

This method returns a tuple (csv_path, query_string). To get just the CSV path: csv_path, _ = builder.to_csv_file()

to_dataframe() → tuple[DataFrame, str]

Get query string and result in pandas.DataFrame.

Returns:

A tuple containing:

• pd.DataFrame: The pandas DataFrame object

• str: The query string executed

Return type:

tuple

Note

This method returns a tuple (dataframe, query_string). To get just the DataFrame: df, _ = builder.to_dataframe()

with_event_time_range(starting_timestamp: datetime | None = None, ending_timestamp: datetime | None = None) → DatasetBuilder

Set event_time_starting_timestamp and event_time_ending_timestamp with provided inputs.

Parameters:

• starting_timestamp (datetime.datetime) – A datetime that all records’ event time in dataset will be after it (default: None).

• ending_timestamp (datetime.datetime) – A datetime that all records’ event time in dataset will be before it (default: None).

Returns:

This DatasetBuilder object.

with_feature_group(feature_group: FeatureGroup, target_feature_name_in_base: str | None = None, included_feature_names: List[str] | None = None, feature_name_in_target: str | None = None, join_comparator: JoinComparatorEnum = JoinComparatorEnum.EQUALS, join_type: JoinTypeEnum = JoinTypeEnum.INNER_JOIN) → DatasetBuilder

Join FeatureGroup with base.

Parameters:

• feature_group (FeatureGroup) – A target FeatureGroup which will be joined to base.

• target_feature_name_in_base (str) – A string representing the feature name in base which will be used as a join key (default: None).

• included_feature_names (List[str]) – A list of strings representing features to be included in the output (default: None).

• feature_name_in_target (str) – A string representing the feature name in the target feature group that will be compared to the target feature in the base feature group. If None is provided, the record identifier feature will be used in the SQL join. (default: None).

• join_comparator (JoinComparatorEnum) – A JoinComparatorEnum representing the comparator used when joining the target feature in the base feature group and the feature in the target feature group. (default: JoinComparatorEnum.EQUALS).

• join_type (JoinTypeEnum) – A JoinTypeEnum representing the type of join between the base and target feature groups. (default: JoinTypeEnum.INNER_JOIN).

Returns:

This DatasetBuilder object.

with_number_of_recent_records_by_record_identifier(n: int) → DatasetBuilder

Set number_of_recent_records field with provided input.

Parameters:

n (int) – An int that how many recent records will be returned for each record identifier.

Returns:

This DatasetBuilder object.

with_number_of_records_from_query_results(n: int) → DatasetBuilder

Set number_of_records field with provided input.

Parameters:

n (int) – An int that how many records will be returned.

Returns:

This DatasetBuilder object.

class sagemaker.mlops.feature_store.dataset_builder.FeatureGroupToBeMerged(features: List[str], included_feature_names: List[str], projected_feature_names: List[str], catalog: str, database: str, table_name: str, record_identifier_feature_name: str, event_time_identifier_feature: FeatureDefinition, target_feature_name_in_base: str | None = None, table_type: TableType | None = None, feature_name_in_target: str | None = None, join_comparator: JoinComparatorEnum = JoinComparatorEnum.EQUALS, join_type: JoinTypeEnum = JoinTypeEnum.INNER_JOIN)

Bases: object

FeatureGroup metadata which will be used for SQL join.

This class instantiates a FeatureGroupToBeMerged object that comprises a list of feature names, a list of feature names which will be included in SQL query, a database, an Athena table name, a feature name of record identifier, a feature name of event time identifier and a feature name of base which is the target join key.

features

A list of strings representing feature names of this FeatureGroup.

Type:

List[str]

included_feature_names

A list of strings representing features to be included in the SQL join.

Type:

List[str]

projected_feature_names

A list of strings representing features to be included for final projection in output.

Type:

List[str]

catalog

A string representing the catalog.

Type:

str

database

A string representing the database.

Type:

str

table_name

A string representing the Athena table name of this FeatureGroup.

Type:

str

record_identifier_feature_name

A string representing the record identifier feature.

Type:

str

event_time_identifier_feature

A FeatureDefinition representing the event time identifier feature.

Type:

FeatureDefinition

target_feature_name_in_base

A string representing the feature name in base which will be used as target join key (default: None).

Type:

str

table_type

A TableType representing the type of table if it is Feature Group or Panda Data Frame (default: None).

Type:

TableType

feature_name_in_target

A string representing the feature name in the target feature group that will be compared to the target feature in the base feature group. If None is provided, the record identifier feature will be used in the SQL join. (default: None).

Type:

str

join_comparator

A JoinComparatorEnum representing the comparator used when joining the target feature in the base feature group and the feature in the target feature group. (default: JoinComparatorEnum.EQUALS).

Type:

JoinComparatorEnum

join_type

A JoinTypeEnum representing the type of join between the base and target feature groups. (default: JoinTypeEnum.INNER_JOIN).

Type:

JoinTypeEnum

catalog: str

database: str

event_time_identifier_feature: FeatureDefinition

feature_name_in_target: str = None

features: List[str]

included_feature_names: List[str]

join_comparator: JoinComparatorEnum = '='

join_type: JoinTypeEnum = 'JOIN'

projected_feature_names: List[str]

record_identifier_feature_name: str

table_name: str

table_type: TableType = None

target_feature_name_in_base: str = None

class sagemaker.mlops.feature_store.dataset_builder.JoinComparatorEnum(value)

Bases: Enum

An enumeration.

EQUALS = '='

GREATER_THAN = '>'

GREATER_THAN_OR_EQUAL_TO = '>='

LESS_THAN = '<'

LESS_THAN_OR_EQUAL_TO = '<='

NOT_EQUAL_TO = '<>'

class sagemaker.mlops.feature_store.dataset_builder.JoinTypeEnum(value)

Bases: Enum

An enumeration.

CROSS_JOIN = 'CROSS JOIN'

FULL_JOIN = 'FULL JOIN'

INNER_JOIN = 'JOIN'

LEFT_JOIN = 'LEFT JOIN'

RIGHT_JOIN = 'RIGHT JOIN'

class sagemaker.mlops.feature_store.dataset_builder.TableType(value)

Bases: Enum

An enumeration.

DATA_FRAME = 'DataFrame'

FEATURE_GROUP = 'FeatureGroup'

sagemaker.mlops.feature_store.dataset_builder.construct_feature_group_to_be_merged(target_feature_group: FeatureGroup, included_feature_names: List[str], target_feature_name_in_base: str | None = None, feature_name_in_target: str | None = None, join_comparator: JoinComparatorEnum = JoinComparatorEnum.EQUALS, join_type: JoinTypeEnum = JoinTypeEnum.INNER_JOIN) → FeatureGroupToBeMerged

Construct a FeatureGroupToBeMerged object by provided parameters.

Parameters:

• target_feature_group (FeatureGroup) – A FeatureGroup object.

• included_feature_names (List[str]) – A list of strings representing features to be included in the output.

• target_feature_name_in_base (str) – A string representing the feature name in base which will be used as target join key (default: None).

• feature_name_in_target (str) – A string representing the feature name in the target feature group that will be compared to the target feature in the base feature group. If None is provided, the record identifier feature will be used in the SQL join. (default: None).

• join_comparator (JoinComparatorEnum) – A JoinComparatorEnum representing the comparator used when joining the target feature in the base feature group and the feature in the target feature group. (default: JoinComparatorEnum.EQUALS).

• join_type (JoinTypeEnum) – A JoinTypeEnum representing the type of join between the base and target feature groups. (default: JoinTypeEnum.INNER_JOIN).

Returns:

A FeatureGroupToBeMerged object.

Raises:

• RuntimeError – No metastore is configured with the FeatureGroup.

• ValueError – Invalid feature name(s) in included_feature_names.