Package com.logicalclocks.hsfs.flink

Class StreamFeatureGroup

java.lang.Object

com.logicalclocks.hsfs.FeatureGroupBase<org.apache.flink.streaming.api.datastream.DataStream<?>>

com.logicalclocks.hsfs.flink.StreamFeatureGroup

public class StreamFeatureGroup
extends FeatureGroupBase<org.apache.flink.streaming.api.datastream.DataStream<?>>

Nested Class Summary

Nested Classes

Modifier and Type	Class	Description
static class	StreamFeatureGroup.StreamFeatureGroupBuilder

Field Summary

Fields

Modifier and Type	Field	Description
protected FeatureGroupEngine	featureGroupEngine

Fields inherited from class com.logicalclocks.hsfs.FeatureGroupBase

created, creator, dataSource, deltaStreamerJobConf, deprecated, description, eventTime, expectationsNames, featureGroupEngineBase, features, featureStore, featurestoreId, hudiPrecombineKey, id, location, LOGGER, name, notificationTopicName, onlineConfig, onlineEnabled, onlineIngestionApi, onlineTopicName, partitionKeys, primaryKeys, statisticColumns, statisticsConfig, subject, timeTravelFormat, topicName, type, utils, version

Constructor Summary

Constructors

Constructor	Description
StreamFeatureGroup()
StreamFeatureGroup(FeatureStore featureStore, int id)
StreamFeatureGroup(FeatureStore featureStore, @NonNull String name, Integer version, String description, List<String> primaryKeys, List<String> partitionKeys, String hudiPrecombineKey, boolean onlineEnabled, TimeTravelFormat timeTravelFormat, List<Feature> features, StatisticsConfig statisticsConfig, String onlineTopicName, String topicName, String notificationTopicName, String eventTime, OnlineConfig onlineConfig, StorageConnector storageConnector, String path)
StreamFeatureGroup(Integer id, String description, List<Feature> features)

Method Summary

Modifier and Type	Method	Description
org.apache.flink.streaming.api.datastream.DataStreamSink<?>	insertStream(org.apache.flink.streaming.api.datastream.DataStream<?> featureData)	Ingest a feature data to the online feature store using Flink DataStream API.
org.apache.flink.streaming.api.datastream.DataStreamSink<?>	insertStream(org.apache.flink.streaming.api.datastream.DataStream<?> featureData, Map<String,String> writeOptions)
void	save()	Save the feature group metadata on Hopsworks.
void	save(Map<String,String> writeOptions, JobConfiguration materializationJobConfiguration)	Save the feature group metadata on Hopsworks.
com.logicalclocks.hsfs.flink.constructor.Query	select(List<String> features)	Select a subset of features of the feature group and return a query object.
com.logicalclocks.hsfs.flink.constructor.Query	selectAll()	Select all features of the feature group and return a query object.
com.logicalclocks.hsfs.flink.constructor.Query	selectExcept(List<String> features)	Select all features including primary key and event time feature of the feature group except provided `features` and return a query object.
com.logicalclocks.hsfs.flink.constructor.Query	selectExceptFeatures(List<Feature> features)	Select all features including primary key and event time feature of the feature group except provided `features` and return a query object.
com.logicalclocks.hsfs.flink.constructor.Query	selectFeatures(List<Feature> features)	Select a subset of features of the feature group and return a query object.

Methods inherited from class com.logicalclocks.hsfs.FeatureGroupBase

addTag, checkDeprecated, delete, deleteTag, getAvroSchema, getComplexFeatures, getDeserializedAvroSchema, getDeserializedEncodedAvroSchema, getEncodedAvroSchema, getFeature, getFeatureAvroSchema, getLatestOnlineIngestion, getOnlineIngestion, getPrimaryKeys, getSubject, getTag, getTags, setDataSource, setDeprecated, unloadSubject, updateDeprecated, updateDeprecated, updateDescription, updateFeatureDescription, updateNotificationTopicName, updateStatisticsConfig

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Field Details

featureGroupEngine

protected FeatureGroupEngine featureGroupEngine

Constructor Details

StreamFeatureGroup

public StreamFeatureGroup(FeatureStore featureStore,
 @NonNull
 @NonNull String name,
 Integer version,
 String description,
 List<String> primaryKeys,
 List<String> partitionKeys,
 String hudiPrecombineKey,
 boolean onlineEnabled,
 TimeTravelFormat timeTravelFormat,
 List<Feature> features,
 StatisticsConfig statisticsConfig,
 String onlineTopicName,
 String topicName,
 String notificationTopicName,
 String eventTime,
 OnlineConfig onlineConfig,
 StorageConnector storageConnector,
 String path)

StreamFeatureGroup

public StreamFeatureGroup()

StreamFeatureGroup

public StreamFeatureGroup(Integer id,
 String description,
 List<Feature> features)

StreamFeatureGroup

public StreamFeatureGroup(FeatureStore featureStore,
 int id)

Method Details

save

public void save()
          throws FeatureStoreException,
IOException

Save the feature group metadata on Hopsworks. This method is idempotent, if the feature group already exists the method does nothing.

Throws:

FeatureStoreException

IOException

save

public void save(Map<String,String> writeOptions,
 JobConfiguration materializationJobConfiguration)
          throws FeatureStoreException,
IOException

Save the feature group metadata on Hopsworks. This method is idempotent, if the feature group already exists, the method does nothing

Parameters:

writeOptions - Options to provide to the materialization job

materializationJobConfiguration - Resource configuration for the materialization job

Throws:

FeatureStoreException

IOException

insertStream

public org.apache.flink.streaming.api.datastream.DataStreamSink<?> insertStream(org.apache.flink.streaming.api.datastream.DataStream<?> featureData)
                                                                         throws Exception

Ingest a feature data to the online feature store using Flink DataStream API. Currently, only POJO types as feature data type are supported.

 
        // get feature store handle
        FeatureStore fs = HopsworksConnection.builder().build().getFeatureStore();

        // get feature group handle
        StreamFeatureGroup fg = fs.getStreamFeatureGroup("card_transactions", 1);

        // read stream from the source and aggregate stream
        DataStream<TransactionAgg> aggregationStream =
          env.fromSource(transactionSource, customWatermark, "Transaction Kafka Source")
          .keyBy(r -> r.getCcNum())
          .window(SlidingEventTimeWindows.of(Time.minutes(windowLength), Time.minutes(1)))
          .aggregate(new TransactionCountAggregate());

        // insert streaming feature data
        fg.insertStream(featureData);
 
 

Specified by:

insertStream in class FeatureGroupBase<org.apache.flink.streaming.api.datastream.DataStream<?>>

Parameters:

featureData - Features in Streaming Dataframe to be saved.

Returns:

DataStreamSink object.

Throws:

Exception

insertStream

public org.apache.flink.streaming.api.datastream.DataStreamSink<?> insertStream(org.apache.flink.streaming.api.datastream.DataStream<?> featureData,
 Map<String,String> writeOptions)
                                                                         throws Exception

Specified by:

insertStream in class FeatureGroupBase<org.apache.flink.streaming.api.datastream.DataStream<?>>

Throws:

Exception

selectFeatures

public com.logicalclocks.hsfs.flink.constructor.Query selectFeatures(List<Feature> features)

Select a subset of features of the feature group and return a query object. The query can be used to construct joins of feature groups or create a feature view with a subset of features of the feature group.

Parameters:

features - List of Feature meta data objects.

Returns:

Query object.

select

public com.logicalclocks.hsfs.flink.constructor.Query select(List<String> features)

Select a subset of features of the feature group and return a query object. The query can be used to construct joins of feature groups or create a feature view with a subset of features of the feature group.

Parameters:

features - List of Feature names.

Returns:

Query object.

selectAll

public com.logicalclocks.hsfs.flink.constructor.Query selectAll()

Select all features of the feature group and return a query object. The query can be used to construct joins of feature groups or create a feature view with a subset of features of the feature group.

Returns:

Query object.

selectExceptFeatures

public com.logicalclocks.hsfs.flink.constructor.Query selectExceptFeatures(List<Feature> features)

Select all features including primary key and event time feature of the feature group except provided `features` and return a query object. The query can be used to construct joins of feature groups or create a feature view with a subset of features of the feature group.

Parameters:

features - List of Feature meta data objects.

Returns:

Query object.

selectExcept

public com.logicalclocks.hsfs.flink.constructor.Query selectExcept(List<String> features)

Select all features including primary key and event time feature of the feature group except provided `features` and return a query object. The query can be used to construct joins of feature groups or create a feature view with a subset of features of the feature group.

Parameters:

features - List of Feature names.

Returns:

Query object.