hsfs.core.data_source #

DataSource #

Metadata object used to provide data source information.

You can obtain data sources using FeatureStore.get_data_source.

The DataSource class encapsulates the details of a data source that can be used for reading or writing data. It supports various types of sources, such as SQL queries, database tables, file paths, and storage connectors.

For Google Sheets connectors, construct a DataSource directly and pass it to FeatureStore.create_feature_group with sink_enabled=True:

from hsfs.core.data_source import DataSource

connector = fs.get_storage_connector("my_google_sheets_connector")

fg = fs.create_feature_group(
    name="budget_actuals",
    version=1,
    primary_key=["id"],
    data_source=DataSource(
        storage_connector=connector,
        table="Q1 Actuals",          # sheet tab name (required)
        spreadsheet_id="1BxiMVs…",  # omit if already set on the connector
    ),
    sink_enabled=True,
)
fg.save()
fg.sink_job.run()

PARAMETER	DESCRIPTION
`query`	SQL query string for the data source, if applicable. TYPE: `str \| None` DEFAULT: `None`
`database`	Name of the database containing the data source. TYPE: `str \| None` DEFAULT: `None`
`group`	Group or schema name for the data source. TYPE: `str \| None` DEFAULT: `None`
`table`	Table name for the data source. For Google Sheets connectors this is the sheet tab name. TYPE: `str \| None` DEFAULT: `None`
`path`	File system path for the data source. TYPE: `str \| None` DEFAULT: `None`
`storage_connector`	Storage connector object holds configuration for accessing the data source. TYPE: `sc.StorageConnector \| dict[str, Any] \| None` DEFAULT: `None`
`metrics`	List of metric column names for the data source. TYPE: `list[str] \| None` DEFAULT: `None`
`dimensions`	List of dimension column names for the data source. TYPE: `list[str] \| None` DEFAULT: `None`
`rest_endpoint`	REST endpoint configuration for the data source. TYPE: `RestEndpointConfig \| dict \| None` DEFAULT: `None`
`spreadsheet_id`	Google Spreadsheet ID for Google Sheets data sources. Only required when the spreadsheet ID was not set on the connector itself; the connector-level value is used otherwise. TYPE: `str \| None` DEFAULT: `None`

database `property` `writable` #

database: str | None

Get or set the database name for the data source.

format `property` `writable` #

format: str | None

Get or set the format of the data source.

group `property` `writable` #

group: str | None

Get or set the group/schema name for the data source.

path `property` `writable` #

path: str | None

Get or set the file system path for the data source.

query `property` `writable` #

query: str | None

Get or set the SQL query string for the data source.

spreadsheet_id `property` `writable` #

spreadsheet_id: str | None

Get or set the Google Spreadsheet ID for Google Sheets data sources.

Only required when the spreadsheet ID was not configured on the connector itself. The connector-level value is used when this is not set.

storage_connector `property` `writable` #

storage_connector: sc.StorageConnector | None

Get or set the storage connector for the data source.

table `property` `writable` #

table: str | None

Get or set the table name for the data source.

For Google Sheets connectors this is the sheet tab name.

estimate_ingestion_resources #

estimate_ingestion_resources(
    feature_group: fg.FeatureGroup | None = None,
    *,
    feature_group_id: int | None = None,
    write_mode: str | None = None,
    loading_strategy: str | None = None,
    batch_size: int | None = None,
    sql_source_fetch_chunk_size: int | None = None,
    source_read_workers: int | None = None,
    data_processing_workers: int | None = None,
    max_upload_batch_size_mb: int | None = None,
    sql_table_num_partitions: int | None = None,
    transform_script_path: str | None = None,
    configured_memory_mb: int | None = None,
    configured_cpu_cores: float | None = None,
) -> dict

Estimate the memory and CPU an ingestion into a feature group needs.

Calls the same backend the UI uses to size a DLTHub ingestion job. It derives a recommendation from the feature group's schema and the runtime knobs you pass, so you can set resource_config on a TableIngestionTarget (or the worker resources of a single sink job) before running an ingestion, instead of guessing. Pass the same runtime knobs you intend to run the ingestion with, since a larger batch size or more workers raises the estimate.

Example

fs = ...
data_source = fs.get_data_source("hubspot")
contacts_fg = fs.get_feature_group("contacts", version=1)

estimate = data_source.estimate_ingestion_resources(
    contacts_fg,
    write_mode="MERGE",
    batch_size=200000,
)
print(estimate["recommendedMemoryMb"], estimate["recommendedCpuCores"])

PARAMETER	DESCRIPTION
`feature_group`	The feature group the ingestion writes to; supplies the schema the estimate is based on. TYPE: `fg.FeatureGroup \| None` DEFAULT: `None`
`feature_group_id`	Id of the target feature group, as an alternative to passing `feature_group`. TYPE: `int \| None` DEFAULT: `None`
`write_mode`	Write mode the ingestion will run with (`APPEND` or `MERGE`). TYPE: `str \| None` DEFAULT: `None`
`loading_strategy`	Loading strategy the ingestion will run with. TYPE: `str \| None` DEFAULT: `None`
`batch_size`	Write batch size the ingestion will run with. TYPE: `int \| None` DEFAULT: `None`
`sql_source_fetch_chunk_size`	Source fetch chunk size for SQL sources. TYPE: `int \| None` DEFAULT: `None`
`source_read_workers`	Number of source read workers. TYPE: `int \| None` DEFAULT: `None`
`data_processing_workers`	Number of data processing workers. TYPE: `int \| None` DEFAULT: `None`
`max_upload_batch_size_mb`	Maximum upload batch size in MB. TYPE: `int \| None` DEFAULT: `None`
`sql_table_num_partitions`	Number of read partitions for SQL sources. TYPE: `int \| None` DEFAULT: `None`
`transform_script_path`	Path of a transformation script the ingestion will run. TYPE: `str \| None` DEFAULT: `None`
`configured_memory_mb`	Memory you plan to give the job, to compare against the recommendation. TYPE: `int \| None` DEFAULT: `None`
`configured_cpu_cores`	CPU cores you plan to give the job, to compare against the recommendation. TYPE: `float \| None` DEFAULT: `None`

RETURNS	DESCRIPTION
`dict`	The backend estimation, including `recommendedMemoryMb`, `recommendedCpuCores`, `confidence`, `peakStage`, `reasons`, and `warnings`.

RAISES	DESCRIPTION
`hopsworks.client.exceptions.RestAPIError`	In case the backend encounters an issue.

get_data #

get_data(use_cached: bool = True) -> dsd.DataSourceData

Retrieve the data from the data source.

Example

# connect to the Feature Store
fs = ...

table = fs.get_data_source("test_data_source").get_tables()[0]

data = table.get_data()

PARAMETER	DESCRIPTION
`use_cached`	Whether to use cached data if available. Only supported for CRM, Google Sheets, and REST connectors. Defaults to `True`. TYPE: `bool` DEFAULT: `True`

RETURNS	DESCRIPTION
`dsd.DataSourceData`	An object containing the data retrieved from the data source.

get_databases #

get_databases() -> list[str]

Retrieve the list of available databases.

Example

# connect to the Feature Store
fs = ...

data_source = fs.get_data_source("test_data_source")

databases = data_source.get_databases()

RETURNS	DESCRIPTION
`list[str]`	A list of database names available in the data source.

get_feature_groups #

get_feature_groups() -> list[fg.FeatureGroup]

Get the feature groups using this data source, based on explicit provenance.

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

RETURNS	DESCRIPTION
`list[fg.FeatureGroup]`	List of feature groups.

get_feature_groups_provenance #

get_feature_groups_provenance() -> Links | None

Get the generated feature groups using this data source, 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.

RETURNS	DESCRIPTION
`Links \| None`	The feature groups generated using this data source or `None` if none were created.

RAISES	DESCRIPTION
`hopsworks.client.exceptions.RestAPIError`	In case the backend encounters an issue.

get_metadata #

get_metadata() -> dict

Retrieve metadata information about the data source.

Example

# connect to the Feature Store
fs = ...

table = fs.get_data_source("test_data_source").get_tables()[0]

metadata = table.get_metadata()

RETURNS	DESCRIPTION
`dict`	A dictionary containing metadata about the data source.

get_tables #

get_tables(database: str | None = None) -> list[DataSource]

Retrieve the list of tables from the specified database.

Example

# connect to the Feature Store
fs = ...

data_source = fs.get_data_source("test_data_source")

tables = data_source.get_tables()

PARAMETER	DESCRIPTION
`database`	The name of the database to list tables from. If not provided, the default database is used. TYPE: `str \| None` DEFAULT: `None`

RETURNS	DESCRIPTION
`list[DataSource]`	A list of DataSource objects representing the tables.

get_training_datasets #

get_training_datasets() -> list[TrainingDataset]

Get the training datasets using this data source, based on explicit provenance.

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

RETURNS	DESCRIPTION
`list[TrainingDataset]`	List of training datasets.

get_training_datasets_provenance #

get_training_datasets_provenance() -> Links

Get the generated training datasets using this data source, 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.

RETURNS	DESCRIPTION
`Links`	The training datasets generated using this data source or `None` if none were created.

RAISES	DESCRIPTION
`hopsworks.client.exceptions.RestAPIError`	In case the backend encounters an issue.

infer_metadata #

infer_metadata(
    preview_data: dsd.DataSourceData | None = None,
) -> InferredMetadata

Use platform intelligence to infer feature metadata for this data source.

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.

Example

fs = ...

table = fs.get_data_source("test_data_source").get_tables()[0]

inferred = table.infer_metadata()
for f in inferred.features:
    print(f.original_name, "->", f.new_name, f.type, f.description)
print("primary key:", inferred.suggested_primary_key)
print("event time:", inferred.suggested_event_time)

PARAMETER	DESCRIPTION
`preview_data`	Pre-fetched preview data to skip a server round-trip; if `None`, a preview is fetched via `get_data`. TYPE: `dsd.DataSourceData \| None` DEFAULT: `None`

RETURNS	DESCRIPTION
`InferredMetadata`	An object containing the suggested feature renames, types, descriptions, primary key, and event time.

RAISES	DESCRIPTION
`hopsworks.client.exceptions.PlatformIntelligenceException`	If platform intelligence is not enabled on the cluster, or the LLM call fails.

new_ingestion_job #

new_ingestion_job(
    name: str,
    *,
    table_parallelism: int = 1,
    environment_name: str | None = None,
    transform_script_path: str | None = None,
    write_mode: str | None = None,
    batch_size: int | None = None,
    sql_source_fetch_chunk_size: int | None = None,
    source_read_workers: int | None = None,
    data_processing_workers: int | None = None,
    max_upload_batch_size_mb: int | None = None,
    sql_table_num_partitions: int | None = None,
    schedule_config: JobSchedule | dict | None = None,
) -> MultiTableIngestionJob

Start assembling a multi-table ingestion job for this data source.

Returns an empty MultiTableIngestionJob you attach feature groups to, either by passing it as sink_job when creating each feature group or by calling MultiTableIngestionJob.add_target. Nothing is created on the server until you call MultiTableIngestionJob.save, so the job is built atomically from the full set of targets.

The arguments here are the job-level defaults every target inherits unless it overrides them.

Example

fs = ...
data_source = fs.get_data_source("hubspot")

job = data_source.new_ingestion_job(name="hubspot_ingestion", table_parallelism=2)

fs.get_or_create_feature_group(
    "contacts", version=1, data_source=data_source,
    sink_enabled=True, sink_job=job,
).save()
fs.get_or_create_feature_group(
    "companies", version=1, data_source=data_source,
    sink_enabled=True, sink_job=job,
).save()

job.save()   # creates one job with both feature groups as targets
job.run()

PARAMETER	DESCRIPTION
`name`	Name of the ingestion job to create. TYPE: `str`
`table_parallelism`	How many tables run at the same time; `1` runs them sequentially. TYPE: `int` DEFAULT: `1`
`environment_name`	Python environment the job runs in. TYPE: `str \| None` DEFAULT: `None`
`transform_script_path`	Default transformation script path for targets that do not set their own. TYPE: `str \| None` DEFAULT: `None`
`write_mode`	Default write mode (`APPEND` or `MERGE`) for targets that do not set their own. TYPE: `str \| None` DEFAULT: `None`
`batch_size`	Default write batch size. TYPE: `int \| None` DEFAULT: `None`
`sql_source_fetch_chunk_size`	Default source fetch chunk size for SQL sources. TYPE: `int \| None` DEFAULT: `None`
`source_read_workers`	Default number of source read workers. TYPE: `int \| None` DEFAULT: `None`
`data_processing_workers`	Default number of data processing workers. TYPE: `int \| None` DEFAULT: `None`
`max_upload_batch_size_mb`	Default maximum upload batch size in MB. TYPE: `int \| None` DEFAULT: `None`
`sql_table_num_partitions`	Default number of read partitions for SQL sources. TYPE: `int \| None` DEFAULT: `None`
`schedule_config`	Optional schedule for the job. TYPE: `JobSchedule \| dict \| None` DEFAULT: `None`

RETURNS	DESCRIPTION
`MultiTableIngestionJob`	An empty ingestion job to collect targets on.

hsfs.core.data_source #

DataSource #

database property writable #

format property writable #

group property writable #

path property writable #

query property writable #

spreadsheet_id property writable #

storage_connector property writable #

table property writable #

estimate_ingestion_resources #

get_data #

get_databases #

get_feature_groups #

get_feature_groups_provenance #

get_metadata #

get_tables #

get_training_datasets #

get_training_datasets_provenance #

infer_metadata #

new_ingestion_job #

database `property` `writable` #

format `property` `writable` #

group `property` `writable` #

path `property` `writable` #

query `property` `writable` #

spreadsheet_id `property` `writable` #

storage_connector `property` `writable` #

table `property` `writable` #