com.logicalclocks.hsfs.spark.engine

Class FeatureGroupEngine

java.lang.Object

com.logicalclocks.hsfs.engine.FeatureGroupEngineBase

com.logicalclocks.hsfs.spark.engine.FeatureGroupEngine

public class FeatureGroupEngine
extends FeatureGroupEngineBase

Field Summary

Fields inherited from class com.logicalclocks.hsfs.engine.FeatureGroupEngineBase

featureGroupApi, LOGGER, tagsApi, utils

Constructor Summary

Constructors

Constructor and Description
FeatureGroupEngine()

Method Summary

Modifier and Type	Method and Description
<T extends FeatureGroupBase> void	appendFeatures(FeatureGroupBase featureGroup, List<Feature> features, Class<T> fgClass)
FeatureGroupCommit	commitDelete(FeatureGroupBase featureGroupBase, org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> genericDataset, Map<String,String> writeOptions)
Map<Long,Map<String,String>>	commitDetails(FeatureGroupBase featureGroupBase, Integer limit)
Map<Long,Map<String,String>>	commitDetailsByWallclockTime(FeatureGroupBase featureGroup, String wallclockTime, Integer limit)
ExternalFeatureGroup	getExternalFeatureGroup(FeatureStore featureStore, String fgName, Integer fgVersion)
List<ExternalFeatureGroup>	getExternalFeatureGroups(FeatureStore featureStore, String fgName)
FeatureGroup	getFeatureGroup(FeatureStore featureStore, String fgName, Integer fgVersion)
List<FeatureGroup>	getFeatureGroups(FeatureStore featureStore, String fgName)
FeatureGroup	getOrCreateFeatureGroup(FeatureStore featureStore, String name, Integer version, String description, List<String> primaryKeys, List<String> partitionKeys, String hudiPrecombineKey, boolean onlineEnabled, TimeTravelFormat timeTravelFormat, StatisticsConfig statisticsConfig, String eventTime)
StreamFeatureGroup	getOrCreateStreamFeatureGroup(FeatureStore featureStore, String name, Integer version, String description, List<String> primaryKeys, List<String> partitionKeys, String hudiPrecombineKey, boolean onlineEnabled, StatisticsConfig statisticsConfig, String eventTime)
StreamFeatureGroup	getStreamFeatureGroup(FeatureStore featureStore, String fgName, Integer fgVersion)
List<StreamFeatureGroup>	getStreamFeatureGroups(FeatureStore featureStore, String fgName)
void	insert(ExternalFeatureGroup externalFeatureGroup, org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> featureData, Map<String,String> writeOptions)
void	insert(FeatureGroup featureGroup, org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> featureData, Storage storage, HudiOperationType operation, org.apache.spark.sql.SaveMode saveMode, List<String> partitionKeys, String hudiPrecombineKey, Map<String,String> writeOptions)
void	insert(StreamFeatureGroup streamFeatureGroup, org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> featureData, org.apache.spark.sql.SaveMode saveMode, List<String> partitionKeys, String hudiPrecombineKey, Map<String,String> writeOptions, JobConfiguration jobConfiguration)
org.apache.spark.sql.streaming.StreamingQuery	insertStream(FeatureGroup featureGroup, org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> featureData, String queryName, String outputMode, boolean awaitTermination, Long timeout, String checkpointLocation, List<String> partitionKeys, String hudiPrecombineKey, Map<String,String> writeOptions) Deprecated.
org.apache.spark.sql.streaming.StreamingQuery	insertStream(StreamFeatureGroup streamFeatureGroup, org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> featureData, String queryName, String outputMode, boolean awaitTermination, Long timeout, String checkpointLocation, List<String> partitionKeys, String hudiPrecombineKey, Map<String,String> writeOptions, JobConfiguration jobConfiguration)
FeatureGroup	save(FeatureGroup featureGroup, org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> dataset, List<String> partitionKeys, String hudiPrecombineKey, Map<String,String> writeOptions) Create the metadata and write the data to the online/offline feature store.
StreamFeatureGroup	save(StreamFeatureGroup featureGroup, org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> dataset, List<String> partitionKeys, String hudiPrecombineKey, Map<String,String> writeOptions, JobConfiguration sparkJobConfiguration) Create the metadata and write the data to the online/offline feature store.
void	saveDataframe(FeatureGroup featureGroup, org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> dataset, Storage storage, HudiOperationType operation, Map<String,String> offlineWriteOptions, Map<String,String> onlineWriteOptions, Integer validationId)
ExternalFeatureGroup	saveExternalFeatureGroup(ExternalFeatureGroup externalFeatureGroup)
FeatureGroup	saveFeatureGroupMetaData(FeatureGroup featureGroup, List<String> partitionKeys, String hudiPrecombineKey, org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> featureData, boolean saveEmpty)
StreamFeatureGroup	saveFeatureGroupMetaData(StreamFeatureGroup featureGroup, List<String> partitionKeys, String hudiPrecombineKey, Map<String,String> writeOptions, JobConfiguration sparkJobConfiguration, org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> featureData)

Methods inherited from class com.logicalclocks.hsfs.engine.FeatureGroupEngineBase

addTag, delete, deleteTag, getTag, getTags, saveExtennalFeatureGroupMetaData, updateDescription, updateFeatures, updateStatisticsConfig

Methods inherited from class java.lang.Object

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

Constructor Detail

FeatureGroupEngine

public FeatureGroupEngine()

Method Detail

save

public FeatureGroup save(FeatureGroup featureGroup,
                         org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> dataset,
                         List<String> partitionKeys,
                         String hudiPrecombineKey,
                         Map<String,String> writeOptions)
                  throws FeatureStoreException,
                         IOException,
                         ParseException

Create the metadata and write the data to the online/offline feature store.

Parameters:

featureGroup - Feature Group metadata object.

dataset - Spark DataFrame or RDD.

partitionKeys - A list of feature names to be used as partition key when writing the feature data to the offline storage, defaults to empty list.

hudiPrecombineKey - A feature name to be used as a precombine key for the `TimeTravelFormat.HUDI` feature group. If feature group has `TimeTravelFormat.HUDI` and hudi precombine key was not specified then the first primary key of the feature group will be used as hudi precombine key.

writeOptions - Additional write options as key-value pairs, defaults to empty Map.

Returns:

Feature Group metadata object

Throws:

FeatureStoreException - In case Client is not connected to Hopsworks, unable to identify date format and/or no commit information was found for the feature group;

IOException - Generic IO exception.

ParseException - In case it's unable to parse HUDI and or statistics commit date string to date type.

save

public StreamFeatureGroup save(StreamFeatureGroup featureGroup,
                               org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> dataset,
                               List<String> partitionKeys,
                               String hudiPrecombineKey,
                               Map<String,String> writeOptions,
                               JobConfiguration sparkJobConfiguration)
                        throws FeatureStoreException,
                               IOException,
                               ParseException

Create the metadata and write the data to the online/offline feature store.

Parameters:

featureGroup -

dataset -

partitionKeys -

writeOptions -

sparkJobConfiguration -

Throws:

FeatureStoreException

IOException

ParseException

insert

public void insert(FeatureGroup featureGroup,
                   org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> featureData,
                   Storage storage,
                   HudiOperationType operation,
                   org.apache.spark.sql.SaveMode saveMode,
                   List<String> partitionKeys,
                   String hudiPrecombineKey,
                   Map<String,String> writeOptions)
            throws FeatureStoreException,
                   IOException,
                   ParseException

Throws:

FeatureStoreException

IOException

ParseException

insert

public void insert(StreamFeatureGroup streamFeatureGroup,
                   org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> featureData,
                   org.apache.spark.sql.SaveMode saveMode,
                   List<String> partitionKeys,
                   String hudiPrecombineKey,
                   Map<String,String> writeOptions,
                   JobConfiguration jobConfiguration)
            throws FeatureStoreException,
                   IOException,
                   ParseException

Throws:

FeatureStoreException

IOException

ParseException

insert

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

Throws:

FeatureStoreException

IOException

insertStream

@Deprecated
public org.apache.spark.sql.streaming.StreamingQuery insertStream(FeatureGroup featureGroup,
                                                                              org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> featureData,
                                                                              String queryName,
                                                                              String outputMode,
                                                                              boolean awaitTermination,
                                                                              Long timeout,
                                                                              String checkpointLocation,
                                                                              List<String> partitionKeys,
                                                                              String hudiPrecombineKey,
                                                                              Map<String,String> writeOptions)
                                                                       throws FeatureStoreException,
                                                                              IOException,
                                                                              org.apache.spark.sql.streaming.StreamingQueryException,
                                                                              TimeoutException,
                                                                              ParseException

Deprecated.

Throws:

FeatureStoreException

IOException

org.apache.spark.sql.streaming.StreamingQueryException

TimeoutException

ParseException

insertStream

public org.apache.spark.sql.streaming.StreamingQuery insertStream(StreamFeatureGroup streamFeatureGroup,
                                                                  org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> featureData,
                                                                  String queryName,
                                                                  String outputMode,
                                                                  boolean awaitTermination,
                                                                  Long timeout,
                                                                  String checkpointLocation,
                                                                  List<String> partitionKeys,
                                                                  String hudiPrecombineKey,
                                                                  Map<String,String> writeOptions,
                                                                  JobConfiguration jobConfiguration)

saveDataframe

public void saveDataframe(FeatureGroup featureGroup,
                          org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> dataset,
                          Storage storage,
                          HudiOperationType operation,
                          Map<String,String> offlineWriteOptions,
                          Map<String,String> onlineWriteOptions,
                          Integer validationId)
                   throws IOException,
                          FeatureStoreException,
                          ParseException

Throws:

IOException

FeatureStoreException

ParseException

saveFeatureGroupMetaData

public FeatureGroup saveFeatureGroupMetaData(FeatureGroup featureGroup,
                                             List<String> partitionKeys,
                                             String hudiPrecombineKey,
                                             org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> featureData,
                                             boolean saveEmpty)
                                      throws FeatureStoreException,
                                             IOException,
                                             ParseException

Throws:

FeatureStoreException

IOException

ParseException

saveFeatureGroupMetaData

public StreamFeatureGroup saveFeatureGroupMetaData(StreamFeatureGroup featureGroup,
                                                   List<String> partitionKeys,
                                                   String hudiPrecombineKey,
                                                   Map<String,String> writeOptions,
                                                   JobConfiguration sparkJobConfiguration,
                                                   org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> featureData)
                                            throws FeatureStoreException,
                                                   IOException

Throws:

FeatureStoreException

IOException

getOrCreateFeatureGroup

public FeatureGroup getOrCreateFeatureGroup(FeatureStore featureStore,
                                            String name,
                                            Integer version,
                                            String description,
                                            List<String> primaryKeys,
                                            List<String> partitionKeys,
                                            String hudiPrecombineKey,
                                            boolean onlineEnabled,
                                            TimeTravelFormat timeTravelFormat,
                                            StatisticsConfig statisticsConfig,
                                            String eventTime)
                                     throws IOException,
                                            FeatureStoreException

Throws:

IOException

FeatureStoreException

getFeatureGroup

public FeatureGroup getFeatureGroup(FeatureStore featureStore,
                                    String fgName,
                                    Integer fgVersion)
                             throws IOException,
                                    FeatureStoreException

Throws:

IOException

FeatureStoreException

getFeatureGroups

public List<FeatureGroup> getFeatureGroups(FeatureStore featureStore,
                                           String fgName)
                                    throws FeatureStoreException,
                                           IOException

Throws:

FeatureStoreException

IOException

getOrCreateStreamFeatureGroup

public StreamFeatureGroup getOrCreateStreamFeatureGroup(FeatureStore featureStore,
                                                        String name,
                                                        Integer version,
                                                        String description,
                                                        List<String> primaryKeys,
                                                        List<String> partitionKeys,
                                                        String hudiPrecombineKey,
                                                        boolean onlineEnabled,
                                                        StatisticsConfig statisticsConfig,
                                                        String eventTime)
                                                 throws IOException,
                                                        FeatureStoreException

Throws:

IOException

FeatureStoreException

getStreamFeatureGroup

public StreamFeatureGroup getStreamFeatureGroup(FeatureStore featureStore,
                                                String fgName,
                                                Integer fgVersion)
                                         throws IOException,
                                                FeatureStoreException

Throws:

IOException

FeatureStoreException

getStreamFeatureGroups

public List<StreamFeatureGroup> getStreamFeatureGroups(FeatureStore featureStore,
                                                       String fgName)
                                                throws FeatureStoreException,
                                                       IOException

Throws:

FeatureStoreException

IOException

appendFeatures

public <T extends FeatureGroupBase> void appendFeatures(FeatureGroupBase featureGroup,
                                                        List<Feature> features,
                                                        Class<T> fgClass)
                                                 throws FeatureStoreException,
                                                        IOException,
                                                        ParseException

Throws:

FeatureStoreException

IOException

ParseException

commitDetails

public Map<Long,Map<String,String>> commitDetails(FeatureGroupBase featureGroupBase,
                                                  Integer limit)
                                           throws IOException,
                                                  FeatureStoreException,
                                                  ParseException

Throws:

IOException

FeatureStoreException

ParseException

commitDetailsByWallclockTime

public Map<Long,Map<String,String>> commitDetailsByWallclockTime(FeatureGroupBase featureGroup,
                                                                 String wallclockTime,
                                                                 Integer limit)
                                                          throws IOException,
                                                                 FeatureStoreException,
                                                                 ParseException

Throws:

IOException

FeatureStoreException

ParseException

commitDelete

public FeatureGroupCommit commitDelete(FeatureGroupBase featureGroupBase,
                                       org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> genericDataset,
                                       Map<String,String> writeOptions)
                                throws IOException,
                                       FeatureStoreException,
                                       ParseException

Throws:

IOException

FeatureStoreException

ParseException

saveExternalFeatureGroup

public ExternalFeatureGroup saveExternalFeatureGroup(ExternalFeatureGroup externalFeatureGroup)
                                              throws FeatureStoreException,
                                                     IOException

Throws:

FeatureStoreException

IOException

getExternalFeatureGroups

public List<ExternalFeatureGroup> getExternalFeatureGroups(FeatureStore featureStore,
                                                           String fgName)
                                                    throws FeatureStoreException,
                                                           IOException

Throws:

FeatureStoreException

IOException

getExternalFeatureGroup

public ExternalFeatureGroup getExternalFeatureGroup(FeatureStore featureStore,
                                                    String fgName,
                                                    Integer fgVersion)
                                             throws IOException,
                                                    FeatureStoreException

Throws:

IOException

FeatureStoreException