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<?>>

Field Summary

Fields

Modifier and Type	Field and Description
protected FeatureGroupEngine	featureGroupEngine

Fields inherited from class com.logicalclocks.hsfs.FeatureGroupBase

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

Constructor Summary

Constructors

Constructor and 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, List<Feature> features, TimeTravelFormat timeTravelFormat, StatisticsConfig statisticsConfig, String onlineTopicName, String topicName, String notificationTopicName, String eventTime, StorageConnector storageConnector, String path)
StreamFeatureGroup(Integer id, String description, List<Feature> features)

Method Summary

Modifier and Type	Method and 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.

Methods inherited from class com.logicalclocks.hsfs.FeatureGroupBase

addTag, checkDeprecated, delete, deleteTag, getAvroSchema, getComplexFeatures, getDeserializedAvroSchema, getDeserializedEncodedAvroSchema, getEncodedAvroSchema, getFeature, getFeatureAvroSchema, getPrimaryKeys, getSubject, getTag, getTags, 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 Detail

featureGroupEngine

protected FeatureGroupEngine featureGroupEngine

Constructor Detail

StreamFeatureGroup

public StreamFeatureGroup(FeatureStore featureStore,
                          @NonNull
                          @NonNull String name,
                          Integer version,
                          String description,
                          List<String> primaryKeys,
                          List<String> partitionKeys,
                          String hudiPrecombineKey,
                          boolean onlineEnabled,
                          List<Feature> features,
                          TimeTravelFormat timeTravelFormat,
                          StatisticsConfig statisticsConfig,
                          String onlineTopicName,
                          String topicName,
                          String notificationTopicName,
                          String eventTime,
                          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 Detail

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