hsfs.storage_connector #

Return prepared options to be passed to an external connector library.

Not implemented for this connector type.

Retrieve the data from the data source.

# connect to the Feature Store
fs = ...

sc = fs.get_storage_connector("conn_name")

tables = sc.get_tables("database_name")

data = sc.get_data(tables[0])

Parameters: data_source (DataSource): The data source to retrieve data from. use_cached (bool): Whether to use cached data if available. Only supported for CRM and REST connectors. Defaults to True.

Retrieve the list of available databases.

# connect to the Feature Store
fs = ...

sc = fs.get_storage_connector("conn_name")

databases = sc.get_databases()

Get the feature groups using this storage connector, based on explicit rovenance.

Only the accessible feature groups are returned. For more items use the base method, see get_feature_groups_provenance.

Get the generated feature groups using this storage connector, based on explicit provenance.

These feature groups can be accessible or inaccessible.

Explicit provenance does not track deleted generated feature group links, so deleted will always be empty. For inaccessible feature groups, only a minimal information is returned.

Retrieve metadata information about the data source.

# connect to the Feature Store
fs = ...

sc = fs.get_storage_connector("conn_name")

tables = sc.get_tables("database_name")

metadata = sc.get_metadata(tables[0])

Retrieve the list of tables from the specified database.

# connect to the Feature Store
fs = ...

sc = fs.get_storage_connector("conn_name")

tables = sc.get_tables("database_name")

Get the training datasets using this storage connector, based on explicit provenance.

Only the accessible training datasets are returned. For more items use the base method, get_training_datasets_provenance.

Get the generated training datasets using this storage connector, based on explicit provenance.

These training datasets can be accessible or inaccessible. Explicit provenance does not track deleted generated training dataset links, so deleted will always be empty. For inaccessible training datasets, only a minimal information is returned.

Use platform intelligence to infer feature metadata for a data source table.

Calls the same backend used by the "Infer metadata" button in the UI when creating an external feature group: an LLM proposes per-column renames, Hopsworks types, descriptions, and a suggested primary key and event time.

fs = ...

sc = fs.get_storage_connector("conn_name")

tables = sc.get_tables("database_name")

inferred = sc.infer_metadata(tables[0])

Prepare Spark to use this Storage Connector.

Reads a query or a path into a dataframe using the storage connector.

Note, paths are only supported for object stores like S3, HopsFS and ADLS, while queries are meant for JDBC or databases like Redshift and Snowflake.

Persist this storage connector to the feature store.

import hopsworks

project = hopsworks.login()
fs = project.get_feature_store()

sc = hsfs.storage_connector.S3Connector(
    id=None,
    name="my_s3_connector",
    featurestore_id=fs.id,
    bucket="my-bucket",
    region="eu-north-1",
)
sc.save()

Return prepared options to be passed to Spark, based on the additional arguments.

Update this storage connector in the feature store.

import hopsworks

project = hopsworks.login()
fs = project.get_feature_store()

sc = fs.get_data_source("my_s3_connector").storage_connector
sc._bucket = "new-bucket"
sc.update()

Prepare Spark to use this Storage Connector.

conn.prepare_spark()

spark.read.format("json").load("abfss://[container-name]@[account_name].dfs.core.windows.net/[path]")

# or
spark.read.format("json").load(conn.prepare_spark("abfss://[container-name]@[account_name].dfs.core.windows.net/[path]"))

Reads a path into a dataframe using the storage connector.

Return options to be passed to an external BigQuery connector library.

Reads results from BigQuery into a spark dataframe using the storage connector.

Reading from bigquery is done via either specifying the BigQuery table or BigQuery query. For example, to read from a BigQuery table, set the BigQuery project, dataset and table on storage connector and read directly from the corresponding path.

conn.read()

conn.read(query='SQL')

conn.read(path='project.dataset.table')

Prepare Spark to use this Storage Connector.

conn.prepare_spark()
spark.read.format("json").load("gs://bucket/path")
# or
spark.read.format("json").load(conn.prepare_spark("gs://bucket/path"))

Reads GCS path into a dataframe using the storage connector.

To read directly from the default bucket, you can omit the path argument:

conn.read(data_format='spark_formats')

conn.read(data_format='spark_formats', paths='Path/object')

conn.read(data_format='spark_formats',path='gs://BUCKET/DATA')

Reads a path into a dataframe using the HopsFS storage connector.

Reads a query into a dataframe using the storage connector.

Return prepared options to be passed to confluent_kafka, based on the provided apache spark configuration.

Right now only producer values with Importance >= medium are implemented.

See https://docs.confluent.io/platform/current/clients/librdkafka/html/md_CONFIGURATION.html.

Create PEM (Privacy Enhanced Mail) files for Kafka SSL authentication.

This method writes the necessary PEM files for SSL authentication with Kafka, using the provided keystore and truststore locations and passwords. The generated file paths are stored as the following instance variables:

- self.ca_chain_path: Path to the generated CA chain PEM file.
- self.client_cert_path: Path to the generated client certificate PEM file.
- self.client_key_path: Path to the generated client key PEM file.

These files are used for configuring secure Kafka connections (e.g., with Spark or confluent_kafka). The method is idempotent and will only create the files once per connector instance.

Return prepared options to be passed to kafka, based on the additional arguments.

See https://kafka.apache.org/documentation/.

Reads a Kafka stream from a topic or multiple topics into a Dataframe.

Return prepared options to be passed to Spark, based on the additional arguments.

This is done by just adding 'kafka.' prefix to kafka_options.

See https://spark.apache.org/docs/latest/structured-streaming-kafka-integration.html#kafka-specific-configurations.

Return arguments suitable for an external pymongo client.

from pymongo import MongoClient

sc = fs.get_storage_connector("mongo_conn")
client = MongoClient(**sc.connector_options())

Forwards any persisted self.options whose key looks like a MongoClient constructor kwarg (lowercase letters, digits, and underscores) so operator-set tuning knobs — maxPoolSize, serverSelectionTimeoutMS, tlsAllowInvalidCertificates, etc. — reach the driver. Keys that look like URI parameters (camelCase, already embedded in connection_uri) and anything non-string are dropped to avoid duplicate-config errors from pymongo.

Ensure the Spark session is wired with the mongo-spark-connector classpath.

Read a collection from MongoDB into a dataframe.

Return options to be passed to an external Redshift connector library.

Reads a table or query into a dataframe using the storage connector.

Refetch storage connector in order to retrieve updated temporary credentials.

Return options to be passed to an external S3 connector library.

Prepare Spark to use this Storage Connector.

conn.prepare_spark()

spark.read.format("json").load("s3a://[bucket]/path")

# or
spark.read.format("json").load(conn.prepare_spark("s3a://[bucket]/path"))

Reads a query or a path into a dataframe using the storage connector.

Note, paths are only supported for object stores like S3, HopsFS and ADLS, while queries are meant for JDBC or databases like Redshift and Snowflake.

Return arguments suitable for an external Python HANA driver such as hdbcli.

from hdbcli import dbapi

sc = fs.get_storage_connector("hana_conn")
conn = dbapi.connect(**sc.connector_options())

Prepare the Spark session with the SAP HANA driver classpath when needed.

Read a table or query from SAP HANA into a dataframe.

Prepare a Python dictionary with the needed arguments for you to connect to a Snowflake database.

It is useful for the snowflake.connector Python library.

import snowflake.connector

sc = fs.get_storage_connector("snowflake_conn")
ctx = snowflake.connector.connect(**sc.connector_options())

Reads a table or query into a dataframe using the storage connector.

Alias for connector_options.

Return UC connector options shaped for external library use.

Read a Unity Catalog table as a Spark DataFrame.

Default behavior: if the SparkSession is connected to a Databricks cluster, dispatch to native UC access (spark.read.table()), which is faster, auth'd by the cluster identity, and skips the Hopsworks round-trip. Otherwise resolves vended S3 credentials via Hopsworks and reads the Delta path directly.

Set force_vended=True to skip detection and always use the vended path, useful if the cluster identity lacks the grants the connector's SP has.

Resolve Spark read options for a Unity Catalog table.

Mirrors the Python / Arrow Flight architecture: Hopsworks vends the Databricks bearer; the SDK calls Databricks directly for vended S3 temp-credentials, then builds the per-bucket S3A keys + Delta path.

v1 supports AWS-backed Delta tables only. Non-AWS storage (Azure / GCP) raises here. Credentials live ~1 h; call this close to the action that triggers the read.

Apply the per-bucket S3A keys to spark's Hadoop config.

Per-bucket scope (fs.s3a.bucket.<bucket>.*) means adjacent reads to other buckets in the same SparkSession are unaffected. Subsequent reads of the same bucket will pick up the most-recently-applied credentials — matches AWS STS rotation semantics.

Apply credentials, then issue the Delta read.