Skip to navigation
User Guide: SDKData & Graph

Connector Reference

View as Markdown

The Kumo Predict SDK exposes connector classes for object storage, warehouses, uploaded files, AWS Glue, and admin-managed group connectors. Each connector returns SourceTable objects that you then convert into Kumo Table objects for your graph.

import kumoai as kumo
kumo.init(url="https://<customer_id>.kumoai.cloud/api", api_key="<api_key>")
connector = kumo.S3Connector(root_dir="s3://my-bucket/kumo-input/")
orders_src = connector["orders"]        # same as connector.table("orders")
print(connector.table_names())
print(orders_src.head())

Connector classes

ConnectorUse forRequired setup
S3ConnectorCSV or Parquet files in Amazon S3Kumo deployment role can read the bucket/prefix
GCSConnectorCSV or Parquet files in Google Cloud StorageKumo workload identity can read the bucket/prefix
ADLSConnectorCSV or Parquet files in Azure Data Lake StorageKumo workload identity can read abfs:// or abfss:// paths
SnowflakeConnectorTables and views in a Snowflake schemaSnowflake account, warehouse, database, schema, and credentials unless using native SPCS auth
DatabricksConnectorTables in a Databricks catalogDatabricks host plus warehouse/cluster/catalog and OAuth or PAT credentials
BigQueryConnectorTables in a BigQuery datasetProject, dataset, and service-account credentials
GlueConnectorPartitioned Parquet tables in AWS Glue CatalogGlue catalog metadata and IAM access to the backing S3 data
FileUploadConnectorLocal or remote CSV/Parquet files uploaded into KumoUse upload() before reading the table
GroupConnectorAdmin-managed connectors in RBAC-enabled workspacesSelect a group, then call kumo.list_group_connectors()

If you use Kumo in Snowpark Container Services, SnowflakeConnector is the supported connector type.

Object storage connectors

Object storage connectors share the same access pattern. Create the connector with a root directory and read tables by relative table name. If you omit root_dir, pass the full table path to table().

s3 = kumo.S3Connector(root_dir="s3://my-bucket/kumo/")
gcs = kumo.GCSConnector(root_dir="gs://my-bucket/kumo/")
adls = kumo.ADLSConnector(root_dir="abfss://container@account.dfs.core.windows.net/kumo/")
orders = s3.table("orders")
customers = gcs["customers"]
transactions = adls.table("transactions")
# Full-path access is also supported when no root_dir is configured.
events = kumo.S3Connector().table("s3://my-bucket/other/events/")

Warehouse connectors

Warehouse connectors create or attach to named Kumo connector resources. You can supply credentials directly or set the documented environment variables before constructing the connector.

Snowflake

snowflake = kumo.SnowflakeConnector(
    name="prod_snowflake",
    account="<account>",
    warehouse="COMPUTE_WH",
    database="ANALYTICS",
    schema_name="PUBLIC",
    credentials={"user": "<user>", "password": "<password>"},
)
orders = snowflake.table("ORDERS")

SnowflakeConnector can also use SNOWFLAKE_USER, SNOWFLAKE_PASSWORD, SNOWFLAKE_PRIVATE_KEY, and SNOWFLAKE_PRIVATE_KEY_PASSPHRASE environment variables. In SPCS environments initialized with snowflake_credentials, those credentials are reused when connector credentials are omitted; when initialized from a Snowflake notebook with snowflake_application, the SDK can create a native Snowflake connector without explicit connector credentials.

Databricks

databricks = kumo.DatabricksConnector(
    name="prod_databricks",
    host="https://<workspace>.cloud.databricks.com",
    cluster_id="<cluster_id>",
    warehouse_id="<warehouse_id>",
    catalog="main",
    credentials={"token": "<personal_access_token>"},
)
orders = databricks.table("sales.orders")

Use either a personal access token (token or DATABRICKS_TOKEN) or OAuth client credentials (client_id/client_secret or DATABRICKS_CLIENT_ID/DATABRICKS_CLIENT_SECRET), not both.

BigQuery

bigquery = kumo.BigQueryConnector(
    name="prod_bigquery",
    project_id="my-gcp-project",
    dataset_id="analytics",
    credentials={
        "private_key_id": "<private_key_id>",
        "private_key": "<private_key>",
        "client_id": "<client_id>",
        "client_email": "<client_email>",
        "token_uri": "https://oauth2.googleapis.com/token",
        "auth_uri": "https://accounts.google.com/o/oauth2/auth",
    },
)
orders = bigquery.table("orders")

Equivalent environment variables are BIGQUERY_PRIVATE_KEY_ID, BIGQUERY_PRIVATE_KEY, BIGQUERY_CLIENT_ID, BIGQUERY_CLIENT_EMAIL, BIGQUERY_TOKEN_URI, and BIGQUERY_AUTH_URI.

AWS Glue

glue = kumo.GlueConnector(
    name="prod_glue",
    account="123456789012",
    region="us-west-2",
    database="analytics",
)
orders = glue.table("orders")

GlueConnector targets partitioned Parquet tables registered in the Glue Catalog.

Reusing existing connectors

Retrieve a connector stored in Kumo (whether created in code or through the Kumo UI) using get_by_name().

snowflake = kumo.SnowflakeConnector.get_by_name("prod_snowflake")
databricks = kumo.DatabricksConnector.get_by_name("prod_databricks")
bigquery = kumo.BigQueryConnector.get_by_name("prod_bigquery")
glue = kumo.GlueConnector.get_by_name("prod_glue")

Uploaded files

Use FileUploadConnector when you have a local file or a remote file/directory that should be uploaded into Kumo-managed storage before it is used as a source table.

parquet_uploads = kumo.FileUploadConnector(file_type="parquet")
parquet_uploads.upload(name="customers", path="/data/customers.parquet")
csv_uploads = kumo.FileUploadConnector(file_type="csv")
csv_uploads.upload(name="events", path="gs://my-bucket/staging/events_csv/")
customers = parquet_uploads["customers"]
events = csv_uploads.table("events")

Local files larger than about 1 GB are partitioned automatically. Remote single files must be about 1 GB or smaller; for larger remote data, upload a directory of same-format shards.

Group connectors

In RBAC-enabled workspaces, administrators can expose shared connectors at the group level. Users list and select those connectors instead of creating connectors with raw credentials.

kumo.init(
    url="https://<customer_id>.kumoai.cloud/api",
    api_key="<api_key>",
    group="marketing",
    project="churn-modeling",
)
for connector in kumo.list_group_connectors():
    print(connector.display_name, connector.source_type, connector.scope)
connector = kumo.list_group_connectors()[0]
orders = connector.table("sales.orders")

If your workspace uses project scoping, initialize the SDK with the appropriate group and project before listing group connectors.

After selecting source tables

After you have SourceTable objects from a connector, construct your tables and graph:

orders = kumo.Table.from_source_table(
    source_table=orders_src,
    time_column="created_at",
).infer_metadata()
customers = kumo.Table.from_source_table(
    source_table=customers_src,
    primary_key="customer_id",
).infer_metadata()

Next, refer to Tables and Graphs.