Package com.logicalclocks.hsfs.spark

Class FeatureGroup

java.lang.Object
com.logicalclocks.hsfs.FeatureGroupBase<org.apache.spark.sql.Dataset<org.apache.spark.sql.Row>>
com.logicalclocks.hsfs.spark.FeatureGroup
public class FeatureGroup extends FeatureGroupBase<org.apache.spark.sql.Dataset<org.apache.spark.sql.Row>>

  • Field Details

  • Constructor Details

  • Method Details

    • read

      public org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> read() throws FeatureStoreException, IOException
      Reads the feature group from the offline storage as Spark DataFrame. 
       
        // get feature store handle
        FeatureStore fs = HopsworksConnection.builder().build().getFeatureStore();
        // get feature group handle
        FeatureGroup fg = fs.getFeatureGroup("electricity_prices", 1);
        // read feature group
        fg.read()
      Returns:
      Spark DataFrame containing the feature data.
      Throws:
      FeatureStoreException - In case it cannot run read query on storage and/or no commit information was found for this feature group;
      IOException - Generic IO exception.

    • read

      public org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> read(boolean online) throws FeatureStoreException, IOException
      Reads the feature group from the offline or online storage as Spark DataFrame. 
       
        // get feature store handle
        FeatureStore fs = HopsworksConnection.builder().build().getFeatureStore();
        // get feature group handle
        FeatureGroup fg = fs.getFeatureGroup("electricity_prices", 1);
        // read feature group data from online storage
        fg.read(true)
        // read feature group data from offline storage
        fg.read(false)
      Parameters:
      online - Set `online` to `true` to read from the online storage.
      Returns:
      Spark DataFrame containing the feature data.
      Throws:
      FeatureStoreException - In case it cannot run read query on storage and/or no commit information was found for this feature group;
      IOException - Generic IO exception.

    • read

      public org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> read(Map<String,String> readOptions) throws FeatureStoreException, IOException
      Reads the feature group from the offline storage as Spark DataFrame. 
       
        // get feature store handle
        FeatureStore fs = HopsworksConnection.builder().build().getFeatureStore();
        // get feature group handle
        FeatureGroup fg = fs.getFeatureGroup("electricity_prices", 1);
        // define additional read options (this example applies to HUDI enabled FGs)
        Map<String, String> readOptions = new HashMap<String, String>() {{
                                                  put("hoodie.datasource.read.end.instanttime", "20230401211015")
                                                }};
        // read feature group data
        fg.read(readOptions)
      Parameters:
      readOptions - Additional read options as key/value pairs.
      Returns:
      Spark DataFrame containing the feature data.
      Throws:
      FeatureStoreException - In case it cannot run read query on storage and/or no commit information was found for this feature group.
      IOException - Generic IO exception.

    • read

      public org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> read(boolean online, Map<String,String> readOptions) throws FeatureStoreException, IOException
      Reads the feature group from the offline or online storage as Spark DataFrame. 
       
        // get feature store handle
        FeatureStore fs = HopsworksConnection.builder().build().getFeatureStore();
        // get feature group handle
        FeatureGroup fg = fs.getFeatureGroup("electricity_prices", 1);
        // define additional read options (this example applies to HUDI enabled FGs)
        Map<String, String> readOptions = new HashMap<String, String>() {{
                                                  put("hoodie.datasource.read.end.instanttime", "20230401211015")
                                                }};
        // read feature group data from online storage
        fg.read(true, readOptions)
        // read feature group data from offline storage
        fg.read(false, readOptions)
      Parameters:
      online - Set `online` to `true` to read from the online storage.
      readOptions - Additional read options as key/value pairs.
      Returns:
      Spark DataFrame containing the feature data.
      Throws:
      FeatureStoreException - In case it cannot run read query on storage and/or no commit information was found for this feature group.
      IOException - Generic IO exception.

    • read

      public org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> read(String wallclockTime) throws FeatureStoreException, IOException, ParseException
      Reads Feature group into a dataframe at a specific point in time. 
       
        // get feature store handle
        FeatureStore fs = HopsworksConnection.builder().build().getFeatureStore();
        // get feature group handle
        FeatureGroup fg = fs.getFeatureGroup("electricity_prices", 1);
        // read feature group data as of specific point in time (Hudi commit timestamp).
        fg.read("20230205210923")
      Parameters:
      wallclockTime - Read data as of this point in time. Datetime string. The String should be formatted in one of the following formats `yyyyMMdd`, `yyyyMMddHH`, `yyyyMMddHHmm`, or `yyyyMMddHHmmss`.
      Returns:
      Spark DataFrame containing feature data.
      Throws:
      FeatureStoreException - In case it's unable to identify format of the provided wallclockTime date format
      IOException - Generic IO exception.
      ParseException - In case it's unable to parse provided wallclockTime to date type.

    • read

      public org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> read(String wallclockTime, Map<String,String> readOptions) throws FeatureStoreException, IOException, ParseException
      Reads Feature group into a dataframe at a specific point in time. 
       
        // get feature store handle
        FeatureStore fs = HopsworksConnection.builder().build().getFeatureStore();
        // get feature group handle
        FeatureGroup fg = fs.getFeatureGroup("electricity_prices", 1);
        // define additional read options (this example applies to HUDI enabled FGs)
        Map<String, String> readOptions = new HashMap<String, String>() {{
                                                  put("hoodie.datasource.read.end.instanttime", "20230401211015")
                                                }};
        // read feature group data as of specific point in time (Hudi commit timestamp).
        fg.read("20230205210923", readOptions)
      Parameters:
      wallclockTime - Datetime string. The String should be formatted in one of the following formats `yyyyMMdd`, `yyyyMMddHH`, `yyyyMMddHHmm`, or `yyyyMMddHHmmss`.
      readOptions - Additional read options as key-value pairs.
      Returns:
      Spark DataFrame containing feature data.
      Throws:
      FeatureStoreException - In case it's unable to identify format of the provided wallclockTime date format
      IOException - Generic IO exception.
      ParseException - In case it's unable to parse provided wallclockTime to date type.

    • readChanges

      @Deprecated public org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> readChanges(String wallclockStartTime, String wallclockEndTime) throws FeatureStoreException, IOException, ParseException
      Deprecated.
      Throws:
      FeatureStoreException
      IOException
      ParseException

    • readChanges

      @Deprecated public org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> readChanges(String wallclockStartTime, String wallclockEndTime, Map<String,String> readOptions) throws FeatureStoreException, IOException, ParseException
      Deprecated.
      Throws:
      FeatureStoreException
      IOException
      ParseException

    • asOf

      public Query asOf(String wallclockTime) throws FeatureStoreException, ParseException
      Get Query object to retrieve all features of the group at a point in the past. This method selects all features in the feature group and returns a Query object at the specified point in time. This can then either be read into a Dataframe or used further to perform joins or construct a training dataset. 
       
        // get feature store handle
        FeatureStore fs = HopsworksConnection.builder().build().getFeatureStore();
        // get feature group handle
        FeatureGroup fg = fs.getFeatureGroup("electricity_prices", 1);
        // get query object to retrieve feature group feature data as of
        // specific point in time (Hudi commit timestamp).
        fg.asOf("20230205210923")
      Parameters:
      wallclockTime - Read data as of this point in time. Datetime string. The String should be formatted in one of the following formats `yyyyMMdd`, `yyyyMMddHH`, `yyyyMMddHHmm`, or `yyyyMMddHHmmss`.
      Returns:
      Query. The query object with the applied time travel condition
      Throws:
      FeatureStoreException - In case it's unable to identify format of the provided wallclockTime date format
      ParseException - In case it's unable to parse provided wallclockTime to date type.

    • asOf

      public Query asOf(String wallclockTime, String excludeUntil) throws FeatureStoreException, ParseException
      Get Query object to retrieve all features of the group at a point in the past. This method selects all features in the feature group and returns a Query object at the specified point in time. This can then either be read into a Dataframe or used further to perform joins or construct a training dataset. 
       
        // get feature store handle
        FeatureStore fs = HopsworksConnection.builder().build().getFeatureStore();
        // get feature group handle
        FeatureGroup fg = fs.getFeatureGroup("electricity_prices", 1);
        // get query object to retrieve feature group feature data as of specific point in time "20230205210923"
        // but exclude commits until "20230204073411"  (Hudi commit timestamp).
        fg.asOf("20230205210923", "20230204073411")
      Parameters:
      wallclockTime - Read data as of this point in time. Datetime string. The String should be formatted in one of the following formats `yyyyMMdd`, `yyyyMMddHH`, `yyyyMMddHHmm`, or `yyyyMMddHHmmss`.
      excludeUntil - Exclude commits until this point in time. Datetime string. The String should be formatted in one of the following formats `yyyyMMdd`, `yyyyMMddHH`, `yyyyMMddHHmm`, or `yyyyMMddHHmmss`.
      Returns:
      Query. The query object with the applied time travel condition
      Throws:
      FeatureStoreException - In case it's unable to identify format of the provided wallclockTime date format.
      ParseException - In case it's unable to parse provided wallclockTime to date type.

    • show

      public void show(int numRows) throws FeatureStoreException, IOException
      Show the first `n` rows of the feature group. 
       
        // get feature store handle
        FeatureStore fs = HopsworksConnection.builder().build().getFeatureStore();
        // get feature group handle
        FeatureGroup fg = fs.getFeatureGroup("electricity_prices", 1);
        // show top 5 lines of feature group data.
        fg.show(5);
      Parameters:
      numRows - Number of rows to show.
      Throws:
      FeatureStoreException - In case it cannot run read query on storage and/or no commit information was found for this feature group;
      IOException - Generic IO exception.

    • show

      public void show(int numRows, boolean online) throws FeatureStoreException, IOException
      Show the first `n` rows of the feature group. 
       
        // get feature store handle
        FeatureStore fs = HopsworksConnection.builder().build().getFeatureStore();
        // get feature group handle
        FeatureGroup fg = fs.getFeatureGroup("electricity_prices", 1);
        // show top 5 lines of feature data from online storage.
        fg.show(5, true);
      Parameters:
      numRows - Number of rows to show.
      online - If `true` read from online feature store.
      Throws:
      FeatureStoreException - In case it cannot run read query on storage and/or no commit information was found for this feature group;
      IOException - Generic IO exception.

    • save

      @Deprecated public void save(org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> featureData) throws FeatureStoreException, IOException, ParseException
      Deprecated.
      Throws:
      FeatureStoreException
      IOException
      ParseException

    • save

      @Deprecated public void save(org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> featureData, Map<String,String> writeOptions) throws FeatureStoreException, IOException, ParseException
      Deprecated.
      Throws:
      FeatureStoreException
      IOException
      ParseException

    • insert

      public void insert(org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> featureData) throws IOException, FeatureStoreException, ParseException
      Incrementally insert data to a feature group or overwrite all data contained in the feature group. The `features` dataframe can be a Spark DataFrame or RDD. 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/offline feature store. 
       
        // get feature store handle
        FeatureStore fs = HopsworksConnection.builder().build().getFeatureStore();
        // get feature group handle
        FeatureGroup fg = fs.getFeatureGroup("electricity_prices", 1);
        //insert feature data
        fg.insert(featureData);
      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 HUDI schema.
      ParseException - In case it's unable to parse HUDI commit date string to date type.

    • 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 a feature group or overwrite all data contained in the feature group. The `features` dataframe can be a Spark DataFrame or RDD. 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/offline feature store. 
       
        // get feature store handle
        FeatureStore fs = HopsworksConnection.builder().build().getFeatureStore();
        // get feature group handle
        FeatureGroup fg = fs.getFeatureGroup("electricity_prices", 1);
        // Define additional write options (this example applies to HUDI enabled FGs)
        Map<String, String> writeOptions = = new HashMap<String, String>() {{
                           put("hoodie.bulkinsert.shuffle.parallelism", "5");
                           put("hoodie.insert.shuffle.parallelism", "5");
                           put("hoodie.upsert.shuffle.parallelism", "5");}
                           };
        // insert feature 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 HUDI schema.
      ParseException - In case it's unable to parse HUDI commit date string to date type.

    • insert

      public void insert(org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> featureData, Storage storage) throws IOException, FeatureStoreException, ParseException
      Incrementally insert data to a feature group or overwrite all data contained in the feature group. By default, the data is inserted into the offline storage as well as the online storage if the feature group is online enabled. To insert only into the online or offline storage specify Storage.ONLINE or Storage.OFFLINE respectively. The `features` dataframe can be a Spark DataFrame or RDD. If statistics are enabled, statistics are recomputed for the entire feature group. If feature group's time travel format is `HUDI` then `operation` argument can be either `insert` or `upsert`. 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/offline feature store. 
       
        // get feature store handle
        FeatureStore fs = HopsworksConnection.builder().build().getFeatureStore();
        // get feature group handle
        FeatureGroup fg = fs.getFeatureGroup("electricity_prices", 1);
        // insert feature data in offline only
        fg.insert(featureData, Storage.OFFLINE);

        // Or insert feature data in online only
        fg.insert(featureData, Storage.ONLINE);
      Parameters:
      featureData - Spark DataFrame, RDD. Features to be saved.
      storage - Overwrite default behaviour, write to offline storage only with `Storage.OFFLINE` or online only with `Storage.ONLINE`
      Throws:
      IOException - Generic IO exception.
      FeatureStoreException - If client is not connected to Hopsworks; cannot run read query on storage and/or can't reconcile HUDI schema.
      ParseException - In case it's unable to parse HUDI commit date string to date type.

    • insert

      public void insert(org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> featureData, boolean overwrite) throws IOException, FeatureStoreException, ParseException
      Incrementally insert data to a feature group or overwrite all data contained in the feature group. By default, the data is inserted into the offline storage as well as the online storage if the feature group is online enabled. To insert only into the online or offline storage specify Storage.ONLINE or Storage.OFFLINE respectively. The `features` dataframe can be a Spark DataFrame or RDD. If statistics are enabled, statistics are recomputed for the entire feature group. If feature group's time travel format is `HUDI` then `operation` argument can be either `insert` or `upsert`. 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/offline feature store. 
       
        // get feature store handle
        FeatureStore fs = HopsworksConnection.builder().build().getFeatureStore();
        // get feature group handle
        FeatureGroup fg = fs.getFeatureGroup("electricity_prices", 1);
        // insert feature data and drop all data in the feature group before inserting new data
        fg.insert(featureData, true);
      Parameters:
      featureData - Spark DataFrame, RDD. Features to be saved.
      overwrite - Drop all data in the feature group before inserting new data. This does not affect metadata.
      Throws:
      IOException - Generic IO exception.
      FeatureStoreException - If client is not connected to Hopsworks; cannot run read query on storage and/or can't reconcile HUDI schema.
      ParseException - In case it's unable to parse HUDI commit date string to date type.

    • insert

      public void insert(org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> featureData, Storage storage, boolean overwrite) throws IOException, FeatureStoreException, ParseException
      Incrementally insert data to a feature group or overwrite all data contained in the feature group. By default, the data is inserted into the offline storage as well as the online storage if the feature group is online enabled. To insert only into the online or offline storage specify Storage.ONLINE or Storage.OFFLINE respectively. The `features` dataframe can be a Spark DataFrame or RDD. If statistics are enabled, statistics are recomputed for the entire feature group. If feature group's time travel format is `HUDI` then `operation` argument can be either `insert` or `upsert`. 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/offline feature store. 
       
        // get feature store handle
        FeatureStore fs = HopsworksConnection.builder().build().getFeatureStore();
        // get feature group handle
        FeatureGroup fg = fs.getFeatureGroup("electricity_prices", 1);
        // insert feature data in offline only and drop all data in the feature group before inserting new data
        fg.insert(featureData, Storage.OFFLINE, true);

        // Or insert feature data in online only and drop all data in the feature group before inserting new data
        fg.insert(featureData, Storage.ONLINE, true);
      Parameters:
      featureData - Spark DataFrame, RDD. Features to be saved.
      storage - Overwrite default behaviour, write to offline storage only with `Storage.OFFLINE` or online only with `Storage.ONLINE`.
      overwrite - Drop all data in the feature group before inserting new data. This does not affect metadata.
      Throws:
      IOException - Generic IO exception.
      FeatureStoreException - If client is not connected to Hopsworks; cannot run read query on storage and/or can't reconcile HUDI schema.
      ParseException - In case it's unable to parse HUDI commit date string to date type.

    • insert

      public void insert(org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> featureData, boolean overwrite, Map<String,String> writeOptions) throws FeatureStoreException, IOException, ParseException
      Incrementally insert data to a feature group or overwrite all data contained in the feature group. By default, the data is inserted into the offline storage as well as the online storage if the feature group is online enabled. To insert only into the online or offline storage specify Storage.ONLINE or Storage.OFFLINE respectively. The `features` dataframe can be a Spark DataFrame or RDD. If statistics are enabled, statistics are recomputed for the entire feature group. If feature group's time travel format is `HUDI` then `operation` argument can be either `insert` or `upsert`. 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/offline feature store. 
       
        // get feature store handle
        FeatureStore fs = HopsworksConnection.builder().build().getFeatureStore();
        // get feature group handle
        FeatureGroup fg = fs.getFeatureGroup("electricity_prices", 1);
        // Define additional write options (this example applies to HUDI enabled FGs)
        Map<String, String> writeOptions = = new HashMap<String, String>() {{
                           put("hoodie.bulkinsert.shuffle.parallelism", "5");
                           put("hoodie.insert.shuffle.parallelism", "5");
                           put("hoodie.upsert.shuffle.parallelism", "5");}
                           };
        // insert feature data and drop all data in the feature group before inserting new data
        fg.insert(featureData, true, writeOptions);
      Parameters:
      featureData - Spark DataFrame, RDD. Features to be saved.
      overwrite - Drop all data in the feature group before inserting new data. This does not affect metadata.
      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 HUDI schema.
      ParseException - In case it's unable to parse HUDI commit date string to date type.

    • insert

      public void insert(org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> featureData, HudiOperationType operation) throws FeatureStoreException, IOException, ParseException
      Incrementally insert data to a feature group or overwrite all data contained in the feature group. By default, the data is inserted into the offline storage as well as the online storage if the feature group is online enabled. To insert only into the online or offline storage specify Storage.ONLINE or Storage.OFFLINE respectively. The `features` dataframe can be a Spark DataFrame or RDD. If statistics are enabled, statistics are recomputed for the entire feature group. If feature group's time travel format is `HUDI` then `operation` argument can be either `insert` or `upsert`. 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/offline feature store. 
       
        // get feature store handle
        FeatureStore fs = HopsworksConnection.builder().build().getFeatureStore();
        // get feature group handle
        FeatureGroup fg = fs.getFeatureGroup("electricity_prices", 1);
        // Define additional write options (this example applies to HUDI enabled FGs)
        Map<String, String> writeOptions = = new HashMap<String, String>() {{
                           put("hoodie.bulkinsert.shuffle.parallelism", "5");
                           put("hoodie.insert.shuffle.parallelism", "5");
                           put("hoodie.upsert.shuffle.parallelism", "5");}
                           };
        // insert feature data
        fg.insert(featureData, HudiOperationType.INSERT);

        // upsert feature data
        fg.insert(featureData, HudiOperationType.UPSERT);
      Parameters:
      featureData - Spark DataFrame, RDD. Features to be saved.
      operation - commit operation type, INSERT or UPSERT.
      Throws:
      IOException - Generic IO exception.
      FeatureStoreException - If client is not connected to Hopsworks; cannot run read query on storage and/or can't reconcile HUDI schema.
      ParseException - In case it's unable to parse HUDI commit date string to date type.

    • insert

      public void insert(org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> featureData, Storage storage, boolean overwrite, HudiOperationType operation, Map<String,String> writeOptions) throws FeatureStoreException, IOException, ParseException
      Incrementally insert data to a feature group or overwrite all data contained in the feature group. By default, the data is inserted into the offline storage as well as the online storage if the feature group is online enabled. To insert only into the online or offline storage specify Storage.ONLINE or Storage.OFFLINE respectively. The `features` dataframe can be a Spark DataFrame or RDD. If statistics are enabled, statistics are recomputed for the entire feature group. If feature group's time travel format is `HUDI` then `operation` argument can be either `insert` or `upsert`. 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/offline feature store. 
       
        // get feature store handle
        FeatureStore fs = HopsworksConnection.builder().build().getFeatureStore();
        // get feature group handle
        FeatureGroup fg = fs.getFeatureGroup("electricity_prices", 1);
        // Define additional write options (this example applies to HUDI enabled FGs)
        Map<String, String> writeOptions = = new HashMap<String, String>() {{
                           put("hoodie.bulkinsert.shuffle.parallelism", "5");
                           put("hoodie.insert.shuffle.parallelism", "5");
                           put("hoodie.upsert.shuffle.parallelism", "5");}
                           };
        // insert feature data in offline only with additional write options and drop all previous data before new
        // data is inserted
        fg.insert(featureData, Storage.OFFLINE, true, HudiOperationType.INSERT, writeOptions);
      Parameters:
      featureData - Spark DataFrame, RDD. Features to be saved.
      storage - Overwrite default behaviour, write to offline storage only with `Storage.OFFLINE` or online only with `Storage.ONLINE`.
      overwrite - Drop all data in the feature group before inserting new data. This does not affect metadata.
      operation - commit operation type, INSERT or UPSERT.
      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 HUDI schema.
      ParseException - In case it's unable to parse HUDI commit date string to date type.

    • insert

      public void insert(org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> featureData, JobConfiguration jobConfiguration) throws FeatureStoreException, IOException, ParseException
      Throws:
      FeatureStoreException
      IOException
      ParseException

    • insert

      public void insert(org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> featureData, boolean overwrite, Map<String,String> writeOptions, JobConfiguration jobConfiguration) throws FeatureStoreException, IOException, ParseException
      Throws:
      FeatureStoreException
      IOException
      ParseException

    • insertStream

      @Deprecated public org.apache.spark.sql.streaming.StreamingQuery insertStream(org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> featureData) throws org.apache.spark.sql.streaming.StreamingQueryException, IOException, FeatureStoreException, TimeoutException, ParseException
      Deprecated.
      insertStream method is deprecated FeatureGroups. Full capability insertStream is available for StreamFeatureGroups.
      insert streaming dataframe in the Feature group.
      Specified by:
      insertStream in class FeatureGroupBase<org.apache.spark.sql.Dataset<org.apache.spark.sql.Row>>
      Parameters:
      featureData - Spark dataframe containing feature data
      Returns:
      StreamingQuery
      Throws:
      org.apache.spark.sql.streaming.StreamingQueryException - StreamingQueryException
      IOException - IOException
      FeatureStoreException - FeatureStoreException
      TimeoutException - TimeoutException
      ParseException - ParseException

    • insertStream

      @Deprecated public org.apache.spark.sql.streaming.StreamingQuery insertStream(org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> featureData, String queryName) throws org.apache.spark.sql.streaming.StreamingQueryException, IOException, FeatureStoreException, TimeoutException, ParseException
      Deprecated.
      Throws:
      org.apache.spark.sql.streaming.StreamingQueryException
      IOException
      FeatureStoreException
      TimeoutException
      ParseException

    • insertStream

      @Deprecated public org.apache.spark.sql.streaming.StreamingQuery insertStream(org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> featureData, Map<String,String> writeOptions) throws FeatureStoreException, IOException, org.apache.spark.sql.streaming.StreamingQueryException, TimeoutException, ParseException
      Deprecated.
      Specified by:
      insertStream in class FeatureGroupBase<org.apache.spark.sql.Dataset<org.apache.spark.sql.Row>>
      Throws:
      FeatureStoreException
      IOException
      org.apache.spark.sql.streaming.StreamingQueryException
      TimeoutException
      ParseException

    • insertStream

      public org.apache.spark.sql.streaming.StreamingQuery insertStream(org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> featureData, String queryName, Map<String,String> writeOptions) throws FeatureStoreException, IOException, org.apache.spark.sql.streaming.StreamingQueryException, TimeoutException, ParseException
      Throws:
      FeatureStoreException
      IOException
      org.apache.spark.sql.streaming.StreamingQueryException
      TimeoutException
      ParseException

    • insertStream

      @Deprecated public org.apache.spark.sql.streaming.StreamingQuery insertStream(org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> featureData, String queryName, String outputMode) throws org.apache.spark.sql.streaming.StreamingQueryException, IOException, FeatureStoreException, TimeoutException, ParseException
      Deprecated.
      Throws:
      org.apache.spark.sql.streaming.StreamingQueryException
      IOException
      FeatureStoreException
      TimeoutException
      ParseException

    • insertStream

      @Deprecated public org.apache.spark.sql.streaming.StreamingQuery insertStream(org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> featureData, String queryName, String outputMode, String checkpointLocation) throws FeatureStoreException, IOException, org.apache.spark.sql.streaming.StreamingQueryException, TimeoutException, ParseException
      Deprecated.
      Throws:
      FeatureStoreException
      IOException
      org.apache.spark.sql.streaming.StreamingQueryException
      TimeoutException
      ParseException

    • insertStream

      @Deprecated public org.apache.spark.sql.streaming.StreamingQuery insertStream(org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> featureData, String queryName, String outputMode, boolean awaitTermination, Long timeout) throws org.apache.spark.sql.streaming.StreamingQueryException, IOException, FeatureStoreException, TimeoutException, ParseException
      Deprecated.
      insertStream method is deprecated FeatureGroups. Full capability insertStream is available for StreamFeatureGroups.
      insert streaming dataframe in the Feature group.
      Parameters:
      featureData - Spark dataframe containing feature data
      queryName - name of spark StreamingQuery
      outputMode - outputMode
      awaitTermination - whether or not to wait for query Termination
      timeout - timeout
      Returns:
      StreamingQuery
      Throws:
      org.apache.spark.sql.streaming.StreamingQueryException - StreamingQueryException
      IOException - IOException
      FeatureStoreException - FeatureStoreException
      TimeoutException - TimeoutException
      ParseException - ParseException

    • insertStream

      @Deprecated public org.apache.spark.sql.streaming.StreamingQuery insertStream(org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> featureData, String queryName, String outputMode, boolean awaitTermination, Long timeout, String checkpointLocation) throws FeatureStoreException, IOException, org.apache.spark.sql.streaming.StreamingQueryException, TimeoutException, ParseException
      Deprecated.
      Throws:
      FeatureStoreException
      IOException
      org.apache.spark.sql.streaming.StreamingQueryException
      TimeoutException
      ParseException

    • insertStream

      @Deprecated public org.apache.spark.sql.streaming.StreamingQuery insertStream(org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> featureData, String queryName, String outputMode, boolean awaitTermination, String checkpointLocation) throws org.apache.spark.sql.streaming.StreamingQueryException, IOException, FeatureStoreException, TimeoutException, ParseException
      Deprecated.
      insertStream method is deprecated FeatureGroups. Full capability insertStream is available for StreamFeatureGroups.
      insert streaming dataframe in the Feature group.
      Parameters:
      featureData - Spark dataframe containing feature data
      queryName - name of spark StreamingQuery
      outputMode - outputMode
      awaitTermination - whether or not to wait for query Termination
      checkpointLocation - path to checkpoint location directory
      Returns:
      StreamingQuery
      Throws:
      org.apache.spark.sql.streaming.StreamingQueryException - StreamingQueryException
      IOException - IOException
      FeatureStoreException - FeatureStoreException
      TimeoutException - TimeoutException
      ParseException - ParseException

    • insertStream

      @Deprecated public org.apache.spark.sql.streaming.StreamingQuery insertStream(org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> featureData, String queryName, String outputMode, boolean awaitTermination, Long timeout, String checkpointLocation, 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 Object insertStream(org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> featureData, String queryName, String outputMode, boolean awaitTermination, Long timeout, String checkpointLocation, Map<String,String> writeOptions, JobConfiguration jobConfiguration) throws FeatureStoreException, IOException, org.apache.spark.sql.streaming.StreamingQueryException, TimeoutException, ParseException
      Throws:
      FeatureStoreException
      IOException
      org.apache.spark.sql.streaming.StreamingQueryException
      TimeoutException
      ParseException

    • removeRows

      public void removeRows(org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> featureData) throws FeatureStoreException, IOException, ParseException
      Drops records present in the provided DataFrame and commits it as update to this Feature group. This method can only be used on time travel enabled feature groups. 
       
        // get feature store handle
        FeatureStore fs = HopsworksConnection.builder().build().getFeatureStore();
        // get feature group handle
        FeatureGroup fg = fs.getFeatureGroup("electricity_prices", 1);
        // Drops records of feature data and commit
        fg.removeRows(featureData);

      When the feature group is online-enabled the records are deleted from the online store as well. Use removeRows(Dataset, Storage) with Storage.OFFLINE to delete offline only.

      Parameters:
      featureData - Spark DataFrame, RDD. Feature data to be deleted.
      Throws:
      FeatureStoreException - If Client is not connected to Hopsworks and/or no commit information was found for this feature group;
      IOException - Generic IO exception.
      ParseException - In case it's unable to parse HUDI commit date string to date type.

    • removeRows

      public void removeRows(org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> featureData, Map<String,String> writeOptions) throws FeatureStoreException, IOException, ParseException
      Drops records present in the provided DataFrame and commits it as update to this Feature group. This method can only be used on time travel enabled feature groups. 
       
        // get feature store handle
        FeatureStore fs = HopsworksConnection.builder().build().getFeatureStore();
        // get feature group handle
        FeatureGroup fg = fs.getFeatureGroup("electricity_prices", 1);
        // Define additional write options (this example applies to HUDI enabled FGs)
        Map<String, String> writeOptions = = new HashMap<String, String>() {{
                           put("hoodie.bulkinsert.shuffle.parallelism", "5");
                           put("hoodie.insert.shuffle.parallelism", "5");
                           put("hoodie.upsert.shuffle.parallelism", "5");}
                           };
        // Drops records of feature data and commit
        fg.removeRows(featureData, writeOptions);

      When the feature group is online-enabled the records are deleted from the online store as well. Use removeRows(Dataset, Map, Storage) with Storage.OFFLINE to delete offline only.

      Parameters:
      featureData - Spark DataFrame, RDD. Feature data to be deleted.
      writeOptions - Additional write options as key-value pairs.
      Throws:
      FeatureStoreException - If Client is not connected to Hopsworks and/or no commit information was found for this feature group;
      IOException - Generic IO exception.
      ParseException - In case it's unable to parse HUDI commit date string to date type.

    • removeRows

      public void removeRows(org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> featureData, Storage storage) throws FeatureStoreException, IOException, ParseException
      Drops records present in the provided DataFrame from the offline table, and optionally from the online store of an online-enabled feature group. This method can only be used on time travel enabled feature groups.

      featureData needs to carry the key columns the offline delete matches on: the primary key, plus the event time and any partition columns when the feature group has them. The online delete matches on the primary key alone and ignores every other column, so a value passed for a non-key feature has no effect on it.

      Parameters:
      featureData - Spark DataFrame, RDD. Feature data to be deleted.
      storage - The storage to delete from. Null follows the feature group: both stores when it is online-enabled, offline alone when it is not. Storage.OFFLINE deletes from the offline table only and Storage.ONLINE from the online store only, as on insert. A single-store delete is not reconciled later: the rows stay in the other store until they are deleted there too.
      Throws:
      FeatureStoreException - If storage is Storage.ONLINE and the feature group is not online-enabled; or on other client/commit errors.
      IOException - Generic IO exception.
      ParseException - In case it's unable to parse HUDI commit date string to date type.

    • removeRows

      public void removeRows(org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> featureData, Map<String,String> writeOptions, Storage storage) throws FeatureStoreException, IOException, ParseException
      Drops records present in the provided DataFrame from the offline table, and optionally from the online store of an online-enabled feature group. This method can only be used on time travel enabled feature groups.

      featureData needs to carry the key columns the offline delete matches on: the primary key, plus the event time and any partition columns when the feature group has them. The online delete matches on the primary key alone and ignores every other column, so a value passed for a non-key feature has no effect on it.

      Parameters:
      featureData - Spark DataFrame, RDD. Feature data to be deleted.
      writeOptions - Additional write options as key-value pairs.
      storage - The storage to delete from. Null follows the feature group: both stores when it is online-enabled, offline alone when it is not. Storage.OFFLINE deletes from the offline table only and Storage.ONLINE from the online store only, as on insert. A single-store delete is not reconciled later: the rows stay in the other store until they are deleted there too.
      Throws:
      FeatureStoreException - If storage is Storage.ONLINE and the feature group is not online-enabled; or on other client/commit errors.
      IOException - Generic IO exception.
      ParseException - In case it's unable to parse HUDI commit date string to date type.

    • commitDeleteRecord

      @Deprecated public void commitDeleteRecord(org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> featureData) throws FeatureStoreException, IOException, ParseException
      Deprecated.
      use removeRows(Dataset) instead.
      Drops records present in the provided DataFrame.
      Parameters:
      featureData - Spark DataFrame, RDD. Feature data to be deleted.
      Throws:
      FeatureStoreException - on client/commit errors.
      IOException - Generic IO exception.
      ParseException - In case it's unable to parse HUDI commit date string to date type.

    • commitDeleteRecord

      @Deprecated public void commitDeleteRecord(org.apache.spark.sql.Dataset<org.apache.spark.sql.Row> featureData, Map<String,String> writeOptions) throws FeatureStoreException, IOException, ParseException
      Deprecated.
      use removeRows(Dataset, Map) instead.
      Drops records present in the provided DataFrame.
      Parameters:
      featureData - Spark DataFrame, RDD. Feature data to be deleted.
      writeOptions - Additional write options as key-value pairs.
      Throws:
      FeatureStoreException - on client/commit errors.
      IOException - Generic IO exception.
      ParseException - In case it's unable to parse HUDI commit date string to date type.

    • commitDetails

      public Map<Long,Map<String,String>> commitDetails() throws IOException, FeatureStoreException, ParseException
      Retrieves commit timeline for this feature group. This method can only be used on time travel enabled feature groups. 
       
        // get feature store handle
        FeatureStore fs = HopsworksConnection.builder().build().getFeatureStore();
        // get feature group handle
        FeatureGroup fg = fs.getFeatureGroup("electricity_prices", 1);
        // get commit timeline.
        fg.commitDetails();
      Returns:
      commit details.
      Throws:
      FeatureStoreException - If Client is not connected to Hopsworks and/or no commit information was found for this feature group;
      IOException - Generic IO exception.
      ParseException - In case it's unable to parse HUDI commit date string to date type.

    • commitDetails

      public Map<Long,Map<String,String>> commitDetails(Integer limit) throws IOException, FeatureStoreException, ParseException
      Retrieves commit timeline for this feature group. This method can only be used on time travel enabled feature groups. 
       
        // get feature store handle
        FeatureStore fs = HopsworksConnection.builder().build().getFeatureStore();
        // get feature group handle
        FeatureGroup fg = fs.getFeatureGroup("electricity_prices", 1);
        // get latest 10 commit details.
        fg.commitDetails(10);
      Parameters:
      limit - number of commits to return.
      Returns:
      commit details.
      Throws:
      FeatureStoreException - If Client is not connected to Hopsworks and/or no commit information was found for this feature group;
      IOException - Generic IO exception.
      ParseException - In case it's unable to parse HUDI commit date string to date type.

    • commitDetails

      public Map<Long,Map<String,String>> commitDetails(String wallclockTime) throws IOException, FeatureStoreException, ParseException
      Return commit details as of specific point in time. 
       
        // get feature store handle
        FeatureStore fs = HopsworksConnection.builder().build().getFeatureStore();
        // get feature group handle
        FeatureGroup fg = fs.getFeatureGroup("electricity_prices", 1);
        // get commit details as of 20230206
        fg.commitDetails("20230206");
      Parameters:
      wallclockTime - Datetime string. The String should be formatted in one of the following formats `yyyyMMdd`, `yyyyMMddHH`, `yyyyMMddHHmm`, or `yyyyMMddHHmmss`.
      Returns:
      commit details.
      Throws:
      FeatureStoreException - If Client is not connected to Hopsworks, unable to identify format of the provided wallclockTime date format and/or no commit information was found for this feature group;
      IOException - Generic IO exception.
      ParseException - In case it's unable to parse HUDI commit date string to date type.

    • commitDetails

      public Map<Long,Map<String,String>> commitDetails(String wallclockTime, Integer limit) throws IOException, FeatureStoreException, ParseException
      Return commit details as of specific point in time. 
       
        // get feature store handle
        FeatureStore fs = HopsworksConnection.builder().build().getFeatureStore();
        // get feature group handle
        FeatureGroup fg = fs.getFeatureGroup("electricity_prices", 1);
        // get top 10 commit details as of 20230206
        fg.commitDetails("20230206", 10);
      Parameters:
      wallclockTime - Datetime string. The String should be formatted in one of the following formats `yyyyMMdd`, `yyyyMMddHH`, `yyyyMMddHHmm`, or `yyyyMMddHHmmss`.
      limit - number of commits to return.
      Returns:
      commit details.
      Throws:
      FeatureStoreException - If Client is not connected to Hopsworks, unable to identify format of the provided wallclockTime date format and/or no commit information was found for this feature group;
      IOException - Generic IO exception.
      ParseException - In case it's unable to parse HUDI commit date string to date type.

    • selectFeatures

      public Query selectFeatures(List<Feature> features)
      Select a subset of features of the feature group and return a query object. The query can be used to construct joins of feature groups or create a feature view with a subset of features of the feature group.
      Parameters:
      features - List of Feature meta data objects.
      Returns:
      Query object.

    • select

      public Query select(List<String> features)
      Select a subset of features of the feature group and return a query object. The query can be used to construct joins of feature groups or create a feature view with a subset of features of the feature group.
      Parameters:
      features - List of Feature names.
      Returns:
      Query object.

    • selectAll

      public Query selectAll()
      Select all features of the feature group and return a query object. The query can be used to construct joins of feature groups or create a feature view with a subset of features of the feature group.
      Returns:
      Query object.

    • selectExceptFeatures

      public Query selectExceptFeatures(List<Feature> features)
      Select all features including primary key and event time feature of the feature group except provided `features` and return a query object. The query can be used to construct joins of feature groups or create a feature view with a subset of features of the feature group.
      Parameters:
      features - List of Feature meta data objects.
      Returns:
      Query object.

    • selectExcept

      public Query selectExcept(List<String> features)
      Select all features including primary key and event time feature of the feature group except provided `features` and return a query object. The query can be used to construct joins of feature groups or create a feature view with a subset of features of the feature group.
      Parameters:
      features - List of Feature names.
      Returns:
      Query object.

    • updateFeatures

      public void updateFeatures(List<Feature> features) throws FeatureStoreException, IOException, ParseException
      Update the metadata of multiple features. Currently only feature description updates are supported.
      Parameters:
      features - List of Feature metadata objects
      Throws:
      FeatureStoreException - If Client is not connected to Hopsworks, unable to identify date format and/or no commit information was found for this feature group;
      IOException - Generic IO exception.
      ParseException - In case it's unable to parse date string to date type.

    • updateFeatures

      public void updateFeatures(Feature feature) throws FeatureStoreException, IOException, ParseException
      Update the metadata of feature. Currently only feature description updates are supported.
      Parameters:
      feature - Feature metadata object
      Throws:
      FeatureStoreException - If Client is not connected to Hopsworks, unable to identify date format and/or no commit information was found for this feature group;
      IOException - Generic IO exception.
      ParseException - In case it's unable to parse date string to date type.

    • appendFeatures

      public void appendFeatures(List<Feature> features) throws FeatureStoreException, IOException, ParseException
      Append features to the schema of the feature group. It is only possible to append features to a feature group. Removing features is considered a breaking change.
      Parameters:
      features - list of Feature metadata objects
      Throws:
      FeatureStoreException - If Client is not connected to Hopsworks, unable to identify date format and/or no commit information was found for this feature group;
      IOException - Generic IO exception.
      ParseException - In case it's unable to parse date string to date type.

    • appendFeatures

      public void appendFeatures(Feature features) throws FeatureStoreException, IOException, ParseException
      Append a single feature to the schema of the feature group. It is only possible to append features to a feature group. Removing features is considered a breaking change.
      Parameters:
      features - List of Feature metadata objects
      Throws:
      FeatureStoreException - If Client is not connected to Hopsworks, unable to identify date format and/or no commit information was found for this feature group;
      IOException - Generic IO exception.
      ParseException - In case it's unable to parse date string to date type.

    • computeStatistics

      public Statistics computeStatistics() throws FeatureStoreException, IOException
      Recompute the statistics for the feature group and save them to the feature store.
      Returns:
      statistics object of computed statistics
      Throws:
      FeatureStoreException - If Client is not connected to Hopsworks,
      IOException - Generic IO exception.

    • computeStatistics

      public Statistics computeStatistics(String wallclockTime) throws FeatureStoreException, IOException, ParseException
      Recompute the statistics for the feature group and save them to the feature store. Statistics are only computed for data in the offline storage of the feature group. 
       
        // get feature store handle
        FeatureStore fs = HopsworksConnection.builder().build().getFeatureStore();
        // get feature group handle
        FeatureGroup fg = fs.getFeatureGroup("electricity_prices", 1);
        // compute statistics as of 20230206
        fg.computeStatistics("20230206", 10);
      Parameters:
      wallclockTime - Datetime string. The String should be formatted in one of the following formats `yyyyMMdd`, `yyyyMMddHH`, `yyyyMMddHHmm`, or `yyyyMMddHHmmss`.
      Returns:
      statistics object of computed statistics
      Throws:
      FeatureStoreException - In case Client is not connected to Hopsworks, unable to identify format of the provided wallclockTime date format and/or no commit information was found for this feature group;
      IOException - Generic IO exception.
      ParseException - In case it's unable to parse HUDI and or statistics commit date string to date type.

    • getStatistics

      public Statistics getStatistics() throws FeatureStoreException, IOException
      Get the last statistics commit for the feature group.
      Returns:
      statistics object of latest commit
      Throws:
      FeatureStoreException - In case Client is not connected to Hopsworks, unable to identify format of the provided wallclockTime date format and/or no commit information was found for this feature group;
      IOException - Generic IO exception.