Package 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	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)
void	deleteOnlineDataframe(FeatureGroupBase featureGroupBase, org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> dataset, Map<String,String> writeOptions)	Produces an online delete tombstone for every row in the dataset.
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)
String	profile(org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> df, List<String> restrictToColumns, Boolean correlation, Boolean histogram, Boolean exactUniqueness, Boolean kll, Integer histogramBins)
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, Map<String,String> writeOptions)	Writes feature group dataframe to kafka for online-fs ingestion.
void	writeOnlineDataframe(FeatureGroupBase featureGroupBase, org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> dataset, Map<String,String> writeOptions, Storage storage)	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 Details

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,
 Map<String,String> writeOptions)
                          throws FeatureStoreException,
IOException

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

Parameters:

featureGroupBase -

dataset -

writeOptions - options map; supported keys under "online_ingestion_options.*" include "online_ingestion_options.disable_online_ingestion_count" (boolean string), and "online_ingestion_options.upsert_if_newer" (boolean string, only updates a row if the new value is newer than the existing one)

Throws:

FeatureStoreException

IOException

writeOnlineDataframe

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

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

Parameters:

featureGroupBase -

dataset -

writeOptions - options map; see writeOnlineDataframe(FeatureGroupBase, Dataset, Map)

storage - which consumer of the topic is meant to ingest these records: Storage.ONLINE when the offline leg is written straight to the table (or does not exist), so that the offline materialization job does not write the same rows a second time; null when the topic is the only source of both stores and both consumers read the records

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

deleteOnlineDataframe

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

Produces an online delete tombstone for every row in the dataset.

Each message carries the row encoded against the feature group Avro schema plus an operation: delete header; OnlineFS deletes the row by primary key and ignores the values. The dataset needs to carry only the primary key: non-key columns are filled with null so the Avro schema serializes.

The deletes are tracked by the same online ingestion record as an insert, and reported under its DELETED status, so the removed rows are counted apart from written ones.

The tombstone is marked Storage.ONLINE: removeRows applies the offline delete straight to the table, so the offline materialization job has to skip it rather than re-insert the key the delete removed.

Parameters:

featureGroupBase - the online-enabled feature group

dataset - the rows to delete

writeOptions - kafka write options

Throws:

FeatureStoreException

IOException

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,
 Boolean kll,
 Integer histogramBins)

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()