com.logicalclocks.hsfs.spark.engine

Class SparkEngine

java.lang.Object

com.logicalclocks.hsfs.engine.EngineBase

com.logicalclocks.hsfs.spark.engine.SparkEngine

public class SparkEngine
extends EngineBase

Field Summary

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

LOGGER, storageConnectorApi

Method Summary

Modifier and Type	Method and Description
String	addFile(String filePath)
org.apache.spark.sql.Dataset<org.apache.spark.sql.Row>	castColumnType(org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> dataset, List<TrainingDatasetFeature> features)
void	closeSparkSession()
String	constructCheckpointPath(FeatureGroupBase featureGroup, String queryName, String queryPrefix)
org.apache.spark.sql.Dataset<org.apache.spark.sql.Row>	convertToDefaultDataframe(org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> dataset)
<S> S	createEmptyDataFrame(S datasetGeneric)
org.apache.spark.sql.Dataset<org.apache.spark.sql.Row>	encodeComplexFeatures(FeatureGroupBase featureGroupBase, org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> dataset) Encodes all complex type features to binary using their avro type as schema.
String	getCertKey()
static SparkEngine	getInstance()
Map<String,String>	getKafkaConfig(FeatureGroupBase featureGroup, Map<String,String> writeOptions)
String	getKeyStorePath()
Map<String,String>	getReadOptions(Map<String,String> providedOptions, DataFormat dataFormat)
String	getTrustStorePath()
Map<String,String>	getWriteOptions(Map<String,String> providedOptions, DataFormat dataFormat)
protected String	makeQueryName(String queryName, FeatureGroupBase featureGroup)
org.apache.spark.sql.Dataset<org.apache.spark.sql.Row>	objectToDataset(Object obj)
List<Feature>	parseFeatureGroupSchema(org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> dataset, TimeTravelFormat timeTravelFormat)
String	profile(org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> df)
String	profile(org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> df, boolean correlation, boolean histogram)
String	profile(org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> df, List<String> restrictToColumns)
String	profile(org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> df, List<String> restrictToColumns, Boolean correlation, Boolean histogram)
String	profile(org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> df, List<String> restrictToColumns, Boolean correlation, Boolean histogram, Boolean exactUniqueness)
org.apache.spark.sql.Dataset<org.apache.spark.sql.Row>	read(StorageConnector storageConnector, String dataFormat, Map<String,String> readOptions, String location)
org.apache.spark.sql.Dataset<org.apache.spark.sql.Row>	readStream(StorageConnector storageConnector, String dataFormat, String messageFormat, String schema, Map<String,String> options, boolean includeMetadata)
void	registerHudiTemporaryTable(FeatureGroupAlias featureGroupAlias, Map<String,String> readOptions)
org.apache.spark.sql.Dataset<org.apache.spark.sql.Row>	registerOnDemandTemporaryTable(ExternalFeatureGroup onDemandFeatureGroup, String alias)
org.apache.spark.sql.Dataset<org.apache.spark.sql.Row>	sanitizeFeatureNames(org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> dataset)
static void	setInstance(SparkEngine sparkEngine)
void	setupConnectorHadoopConf(StorageConnector storageConnector)
static String	sparkPath(String path)
org.apache.spark.sql.Dataset<org.apache.spark.sql.Row>[]	splitDataset(TrainingDataset trainingDataset, Query query, Map<String,String> readOptions)
static List<org.apache.spark.sql.Dataset<org.apache.spark.sql.Row>>	splitLabels(org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> dataset, List<String> labels)
org.apache.spark.sql.Dataset<org.apache.spark.sql.Row>	sql(String query)
void	streamToHudiTable(StreamFeatureGroup streamFeatureGroup, Map<String,String> writeOptions)
void	validateSparkConfiguration()
org.apache.spark.sql.Dataset<org.apache.spark.sql.Row>[]	write(TrainingDataset trainingDataset, Query query, Map<String,String> queryReadOptions, Map<String,String> writeOptions, org.apache.spark.sql.SaveMode saveMode) Setup Spark to write the data on the File System.
void	writeEmptyDataframe(FeatureGroupBase featureGroup)
void	writeOfflineDataframe(FeatureGroupBase featureGroup, org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> dataset, HudiOperationType operation, Map<String,String> writeOptions, Integer validationId)
void	writeOnlineDataframe(FeatureGroupBase featureGroupBase, org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> dataset, String onlineTopicName, Map<String,String> writeOptions) Writes feature group dataframe to kafka for online-fs ingestion.
<S> org.apache.spark.sql.streaming.StreamingQuery	writeStreamDataframe(FeatureGroupBase featureGroupBase, org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> dataset, String queryName, String outputMode, boolean awaitTermination, Long timeout, String checkpointLocation, Map<String,String> writeOptions)

Methods inherited from class java.lang.Object

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

Method Detail

getInstance

public static SparkEngine getInstance()

setInstance

public static void setInstance(SparkEngine sparkEngine)

validateSparkConfiguration

public void validateSparkConfiguration()
                                throws FeatureStoreException

Throws:

FeatureStoreException

getTrustStorePath

public String getTrustStorePath()

getKeyStorePath

public String getKeyStorePath()

getCertKey

public String getCertKey()

sql

public org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> sql(String query)

registerOnDemandTemporaryTable

public org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> registerOnDemandTemporaryTable(ExternalFeatureGroup onDemandFeatureGroup,
                                                                                             String alias)
                                                                                      throws FeatureStoreException,
                                                                                             IOException

Throws:

FeatureStoreException

IOException

splitLabels

public static List<org.apache.spark.sql.Dataset<org.apache.spark.sql.Row>> splitLabels(org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> dataset,
                                                                                       List<String> labels)

registerHudiTemporaryTable

public void registerHudiTemporaryTable(FeatureGroupAlias featureGroupAlias,
                                       Map<String,String> readOptions)
                                throws FeatureStoreException

Throws:

FeatureStoreException

write

public org.apache.spark.sql.Dataset<org.apache.spark.sql.Row>[] write(TrainingDataset trainingDataset,
                                                                      Query query,
                                                                      Map<String,String> queryReadOptions,
                                                                      Map<String,String> writeOptions,
                                                                      org.apache.spark.sql.SaveMode saveMode)
                                                               throws FeatureStoreException,
                                                                      IOException

Setup Spark to write the data on the File System.

Parameters:

trainingDataset - Training Dataset metadata object

query - Query Object

queryReadOptions - Additional read options as key-value pairs, defaults to empty Map

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

saveMode - org.apache.spark.sql.saveMode: Append, Overwrite, ErrorIfExists, Ignore

Returns:

Spark dataframe

Throws:

FeatureStoreException - If Client is not connected to Hopsworks

IOException - Generic IO exception.

splitDataset

public org.apache.spark.sql.Dataset<org.apache.spark.sql.Row>[] splitDataset(TrainingDataset trainingDataset,
                                                                             Query query,
                                                                             Map<String,String> readOptions)
                                                                      throws FeatureStoreException,
                                                                             IOException

Throws:

FeatureStoreException

IOException

getWriteOptions

public Map<String,String> getWriteOptions(Map<String,String> providedOptions,
                                          DataFormat dataFormat)

getReadOptions

public Map<String,String> getReadOptions(Map<String,String> providedOptions,
                                         DataFormat dataFormat)

read

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

Throws:

FeatureStoreException

IOException

writeOnlineDataframe

public void writeOnlineDataframe(FeatureGroupBase featureGroupBase,
                                 org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> dataset,
                                 String onlineTopicName,
                                 Map<String,String> writeOptions)
                          throws FeatureStoreException,
                                 IOException

Writes feature group dataframe to kafka for online-fs ingestion.

Parameters:

featureGroupBase -

dataset -

writeOptions -

Throws:

FeatureStoreException

IOException

writeStreamDataframe

public <S> org.apache.spark.sql.streaming.StreamingQuery writeStreamDataframe(FeatureGroupBase featureGroupBase,
                                                                              org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> dataset,
                                                                              String queryName,
                                                                              String outputMode,
                                                                              boolean awaitTermination,
                                                                              Long timeout,
                                                                              String checkpointLocation,
                                                                              Map<String,String> writeOptions)
                                                                       throws FeatureStoreException,
                                                                              IOException,
                                                                              org.apache.spark.sql.streaming.StreamingQueryException,
                                                                              TimeoutException

Throws:

FeatureStoreException

IOException

org.apache.spark.sql.streaming.StreamingQueryException

TimeoutException

encodeComplexFeatures

public org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> encodeComplexFeatures(FeatureGroupBase featureGroupBase,
                                                                                    org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> dataset)
                                                                             throws FeatureStoreException,
                                                                                    IOException

Encodes all complex type features to binary using their avro type as schema.

Parameters:

featureGroupBase - FeatureGroupBase Feature Group hsfs metadata object

dataset - Spark DataFrame or RDD.

Returns:

Spark DataFrame.

Throws:

FeatureStoreException - If Client is not connected to Hopsworks

IOException - Generic IO exception.

writeEmptyDataframe

public void writeEmptyDataframe(FeatureGroupBase featureGroup)
                         throws IOException,
                                FeatureStoreException,
                                ParseException

Throws:

IOException

FeatureStoreException

ParseException

writeOfflineDataframe

public void writeOfflineDataframe(FeatureGroupBase featureGroup,
                                  org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> dataset,
                                  HudiOperationType operation,
                                  Map<String,String> writeOptions,
                                  Integer validationId)
                           throws IOException,
                                  FeatureStoreException,
                                  ParseException

Throws:

IOException

FeatureStoreException

ParseException

profile

public String profile(org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> df,
                      List<String> restrictToColumns,
                      Boolean correlation,
                      Boolean histogram,
                      Boolean exactUniqueness)

profile

public String profile(org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> df,
                      List<String> restrictToColumns,
                      Boolean correlation,
                      Boolean histogram)

profile

public String profile(org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> df,
                      List<String> restrictToColumns)

profile

public String profile(org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> df,
                      boolean correlation,
                      boolean histogram)

profile

public String profile(org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> df)

setupConnectorHadoopConf

public void setupConnectorHadoopConf(StorageConnector storageConnector)
                              throws IOException,
                                     FeatureStoreException

Throws:

IOException

FeatureStoreException

sparkPath

public static String sparkPath(String path)

streamToHudiTable

public void streamToHudiTable(StreamFeatureGroup streamFeatureGroup,
                              Map<String,String> writeOptions)
                       throws Exception

Throws:

Exception

parseFeatureGroupSchema

public List<Feature> parseFeatureGroupSchema(org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> dataset,
                                             TimeTravelFormat timeTravelFormat)
                                      throws FeatureStoreException

Throws:

FeatureStoreException

sanitizeFeatureNames

public org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> sanitizeFeatureNames(org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> dataset)

convertToDefaultDataframe

public org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> convertToDefaultDataframe(org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> dataset)

castColumnType

public org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> castColumnType(org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> dataset,
                                                                             List<TrainingDatasetFeature> features)
                                                                      throws FeatureStoreException

Throws:

FeatureStoreException

addFile

public String addFile(String filePath)
               throws FeatureStoreException

Specified by:

addFile in class EngineBase

Throws:

FeatureStoreException

getKafkaConfig

public Map<String,String> getKafkaConfig(FeatureGroupBase featureGroup,
                                         Map<String,String> writeOptions)
                                  throws FeatureStoreException,
                                         IOException

Specified by:

getKafkaConfig in class EngineBase

Throws:

FeatureStoreException

IOException

readStream

public org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> readStream(StorageConnector storageConnector,
                                                                         String dataFormat,
                                                                         String messageFormat,
                                                                         String schema,
                                                                         Map<String,String> options,
                                                                         boolean includeMetadata)
                                                                  throws FeatureStoreException,
                                                                         IOException

Throws:

FeatureStoreException

IOException

objectToDataset

public org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> objectToDataset(Object obj)

createEmptyDataFrame

public <S> S createEmptyDataFrame(S datasetGeneric)

constructCheckpointPath

public String constructCheckpointPath(FeatureGroupBase featureGroup,
                                      String queryName,
                                      String queryPrefix)
                               throws FeatureStoreException,
                                      IOException

Throws:

FeatureStoreException

IOException

makeQueryName

protected String makeQueryName(String queryName,
                               FeatureGroupBase featureGroup)

closeSparkSession

public void closeSparkSession()