com.logicalclocks.hsfs.spark

Class ExternalFeatureGroup

java.lang.Object

com.logicalclocks.hsfs.FeatureGroupBase<org.apache.spark.sql.Dataset<org.apache.spark.sql.Row>>

com.logicalclocks.hsfs.spark.ExternalFeatureGroup

public class ExternalFeatureGroup
extends FeatureGroupBase<org.apache.spark.sql.Dataset<org.apache.spark.sql.Row>>

Field Summary

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
ExternalFeatureGroup()
ExternalFeatureGroup(FeatureStore featureStore, int id)
ExternalFeatureGroup(FeatureStore featureStore, @NonNull String name, Integer version, String query, ExternalDataFormat dataFormat, String path, Map<String,String> options, @NonNull StorageConnector storageConnector, String description, List<String> primaryKeys, List<Feature> features, StatisticsConfig statisticsConfig, String eventTime, boolean onlineEnabled, String onlineTopicName, String topicName, String notificationTopicName)

Method Summary

Modifier and Type	Method and Description
void	appendFeatures(Feature features)
void	appendFeatures(List<Feature> features)
Statistics	computeStatistics()
Statistics	getStatistics()
void	insert(org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> featureData) Incrementally insert data to the online storage of an external feature group.
void	insert(org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> featureData, Map<String,String> writeOptions) Incrementally insert data to the online storage of an external feature group.
void	insert(org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> featureData, Storage storage)
Object	insertStream(org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> featureData)
Object	insertStream(org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> featureData, Map<String,String> writeOptions)
org.apache.spark.sql.Dataset<org.apache.spark.sql.Row>	read()
org.apache.spark.sql.Dataset<org.apache.spark.sql.Row>	read(boolean online)
org.apache.spark.sql.Dataset<org.apache.spark.sql.Row>	read(boolean online, Map<String,String> readOptions)
org.apache.spark.sql.Dataset<org.apache.spark.sql.Row>	read(Map<String,String> readOptions)
void	save()
Query	select(List<String> features)
Query	selectAll()
Query	selectExcept(List<String> features)
Query	selectExceptFeatures(List<Feature> features)
Query	selectFeatures(List<Feature> features)
void	show(int numRows)
void	show(int numRows, boolean online)
void	updateFeatures(Feature feature)
void	updateFeatures(List<Feature> features)

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

Constructor Detail

ExternalFeatureGroup

public ExternalFeatureGroup(FeatureStore featureStore,
                            @NonNull
                            @NonNull String name,
                            Integer version,
                            String query,
                            ExternalDataFormat dataFormat,
                            String path,
                            Map<String,String> options,
                            @NonNull
                            @NonNull StorageConnector storageConnector,
                            String description,
                            List<String> primaryKeys,
                            List<Feature> features,
                            StatisticsConfig statisticsConfig,
                            String eventTime,
                            boolean onlineEnabled,
                            String onlineTopicName,
                            String topicName,
                            String notificationTopicName)

ExternalFeatureGroup

public ExternalFeatureGroup()

ExternalFeatureGroup

public ExternalFeatureGroup(FeatureStore featureStore,
                            int id)

Method Detail

save

public void save()
          throws FeatureStoreException,
                 IOException

Throws:

FeatureStoreException

IOException

read

public org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> read()
                                                            throws FeatureStoreException,
                                                                   IOException

Throws:

FeatureStoreException

IOException

read

public org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> read(boolean online)
                                                            throws FeatureStoreException,
                                                                   IOException

Throws:

FeatureStoreException

IOException

read

public org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> read(Map<String,String> readOptions)
                                                            throws FeatureStoreException,
                                                                   IOException

Throws:

FeatureStoreException

IOException

read

public org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> read(boolean online,
                                                                   Map<String,String> readOptions)
                                                            throws FeatureStoreException,
                                                                   IOException

Throws:

FeatureStoreException

IOException

show

public void show(int numRows)
          throws FeatureStoreException,
                 IOException

Throws:

FeatureStoreException

IOException

show

public void show(int numRows,
                 boolean online)
          throws FeatureStoreException,
                 IOException

Throws:

FeatureStoreException

IOException

insert

public void insert(org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> featureData,
                   Storage storage)
            throws IOException,
                   FeatureStoreException,
                   ParseException

Throws:

IOException

FeatureStoreException

ParseException

insert

public void insert(org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> featureData)
            throws FeatureStoreException,
                   IOException

Incrementally insert data to the online storage of an external feature group. The feature group has to be online enabled to perform this operation. The `features` dataframe can be a Spark DataFrame or RDD. If statistics are enabled, statistics are recomputed for the entire feature group. If the feature group doesn't exist, the insert method will create the necessary metadata the first time it is invoked and write the specified `features` dataframe as feature group to the online feature store.

 
        // get feature store handle
        FeatureStore fs = HopsworksConnection.builder().build().getFeatureStore();
        // get feature group handle
        ExternalFeatureGroup fg = fs.getExternalFeatureGroup("electricity_prices", 1);
        // insert data
        fg.insert(featureData, writeOptions);
 
 

Parameters:

featureData - Spark DataFrame, RDD. Features to be saved.

Throws:

IOException - Generic IO exception.

FeatureStoreException - If client is not connected to Hopsworks; cannot run read query on storage and/or can't reconcile schema.

insert

public void insert(org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> featureData,
                   Map<String,String> writeOptions)
            throws FeatureStoreException,
                   IOException,
                   ParseException

Incrementally insert data to the online storage of an external feature group. The feature group has to be online enabled to perform this operation. The `features` dataframe can be a Spark DataFrame or RDD. If statistics are enabled, statistics are recomputed for the entire feature group. If the feature group doesn't exist, the insert method will create the necessary metadata the first time it is invoked and write the specified `features` dataframe as feature group to the online feature store.

 
        // get feature store handle
        FeatureStore fs = HopsworksConnection.builder().build().getFeatureStore();
        // get feature group handle
        ExternalFeatureGroup fg = fs.getExternalFeatureGroup("electricity_prices", 1);
        // Define additional write options (for example for Spark)
        Map<String, String> writeOptions = = new HashMap<String, String>();
        // insert data
        fg.insert(featureData, writeOptions);
 
 

Parameters:

featureData - Spark DataFrame, RDD. Features to be saved.

writeOptions - Additional write options as key-value pairs.

Throws:

IOException - Generic IO exception.

FeatureStoreException - If client is not connected to Hopsworks; cannot run read query on storage and/or can't reconcile schema.

ParseException

selectFeatures

public Query selectFeatures(List<Feature> features)

select

public Query select(List<String> features)

selectAll

public Query selectAll()

selectExceptFeatures

public Query selectExceptFeatures(List<Feature> features)

selectExcept

public Query selectExcept(List<String> features)

insertStream

public Object insertStream(org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> featureData)
                    throws Exception

Specified by:

insertStream in class FeatureGroupBase<org.apache.spark.sql.Dataset<org.apache.spark.sql.Row>>

Throws:

Exception

insertStream

public Object insertStream(org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> featureData,
                           Map<String,String> writeOptions)
                    throws Exception

Specified by:

insertStream in class FeatureGroupBase<org.apache.spark.sql.Dataset<org.apache.spark.sql.Row>>

Throws:

Exception

updateFeatures

public void updateFeatures(List<Feature> features)
                    throws FeatureStoreException,
                           IOException,
                           ParseException

Throws:

FeatureStoreException

IOException

ParseException

updateFeatures

public void updateFeatures(Feature feature)
                    throws FeatureStoreException,
                           IOException,
                           ParseException

Throws:

FeatureStoreException

IOException

ParseException

appendFeatures

public void appendFeatures(List<Feature> features)
                    throws FeatureStoreException,
                           IOException,
                           ParseException

Throws:

FeatureStoreException

IOException

ParseException

appendFeatures

public void appendFeatures(Feature features)
                    throws FeatureStoreException,
                           IOException,
                           ParseException

Throws:

FeatureStoreException

IOException

ParseException

computeStatistics

public Statistics computeStatistics()
                             throws FeatureStoreException,
                                    IOException

Throws:

FeatureStoreException

IOException

getStatistics

public Statistics getStatistics()
                         throws FeatureStoreException,
                                IOException

Throws:

FeatureStoreException

IOException