ml-analytics-tools 0.8.0

Tools for ML projects and data management

ML Analytics Tools

Utilities for common analytics and machine learning workflows: Redshift, S3, Google Sheets, Slack, MLflow, model evaluation, and SQL pipelines.

The package is intentionally infrastructure-neutral. Buckets, credentials, MLflow hosts, and tokens are provided by your environment or by explicit arguments.

What Is Included

• DataConnector: run Redshift or Snowflake SQL, load SQL files, unload/load data through S3, and create Redshift tables from DataFrames.
• SFConnector: read Snowflake through Spark and save results to Unity Catalog tables (Databricks). PySpark is imported lazily, so the rest of the package works without it.
• SparkTableManager: source-agnostic Spark / Unity Catalog operations — save a Spark/pandas/polars DataFrame to a Delta table, convert pandas/polars to Spark, run Spark SQL, optimize, comment, drop, or read tables.
• S3Connector: read, write, list, delete, and query S3 data with DuckDB.
• GSheet: read, write, share, and export Google Sheets data.
• SlackConnector: send messages, upload files, and manage simple Slack interactions.
• ModelManager: create MLflow experiments, log models, register versions, manage aliases, and handle permissions.
• model_tools: classification, regression, survival analysis, CatBoost helpers, plotting, and reporting utilities.
• utils: project-root discovery, SQL file loading, logging, credentials, and YAML SQL pipelines.

Install

From PyPI, after a release is available:

uv add ml-analytics-tools

Directly from GitHub:

uv add git+https://github.com/sdaza/ml-analytics-tools

For local development:

uv sync --all-groups

Configuration

The package loads a .env file from the project root when it is imported. Only configure the services you use.

# Redshift
BI_REDSHIFT_HOST=redshift-cluster.example.com
BI_REDSHIFT_DB=analytics
BI_REDSHIFT_USER=analytics_user
BI_REDSHIFT_PASSWORD=secret
BI_REDSHIFT_PORT=5439

# Snowflake
SNOWFLAKE_USER=[e-mail]
SNOWFLAKE_ACCOUNT=your-account
SNOWFLAKE_WAREHOUSE=ANALYTICS_S_WH
SNOWFLAKE_DATABASE=ANALYTICS_DB
SNOWFLAKE_SCHEMA=PUBLIC
SNOWFLAKE_ROLE=SNOWFLAKE_STND_DATA
SNOWFLAKE_AUTHENTICATOR=externalbrowser

# Or select a named profile from ~/.snowflake/connections.toml
SNOWFLAKE_CONNECTION_NAME=preply_entp

# Browser-free Snowflake auth for local or Databricks jobs
SNOWFLAKE_PRIVATE_KEY_PATH=~/.snowflake/rsa_key.p8
SNOWFLAKE_PRIVATE_KEY_PASSPHRASE=secret

# S3
ML_ANALYTICS_S3_BUCKET=my-analytics-bucket

# MLflow
MLFLOW_TRACKING_URI=https://mlflow.example.com
MLFLOW_TRACKING_USERNAME=[e-mail]
MLFLOW_TRACKING_PASSWORD=secret

# Google Sheets
GSHEET_SPREADSHEET_ID=optional-default-sheet-id
GOOGLE_CREDENTIALS='{"type":"service_account", ...}'

# Slack
SLACK_BOT_TOKEN=xoxb-your-token

S3 buckets are never hard-coded. Pass bucket=... or s3_bucket=..., or set ML_ANALYTICS_S3_BUCKET.

AWS Authentication

Use the CLI helper for AWS SSO:

ml-analytics-auth

You can also call it from Python:

from ml_analytics import ensure_aws_authenticated

ensure_aws_authenticated()

See AWS Authentication and CLI Commands for details.

Quick Examples

Query Redshift

from ml_analytics import DataConnector

dc = DataConnector()

df = dc.sql("SELECT * FROM analytics.customer_features LIMIT 100")
df_polars = dc.sql("queries/features.sql", format="polars", country="es")

Query Snowflake

from ml_analytics import DataConnector

dc = DataConnector(engine="snowflake")

df = dc.sql("SELECT 1 AS col_1")

Named connection profiles (connection_name)

Instead of SNOWFLAKE_* env vars, you can use a named profile from Snowflake's standard ~/.snowflake/connections.toml / config.toml. The engine is then implicitly snowflake, profile parsing is delegated to the Snowflake connector, and any constructor args act as explicit overrides on top of the profile:

dc = DataConnector(connection_name="preply_entp", role="SNOWFLAKE_ENTP_ANALYST_PII")
# equivalent:
dc = DataConnector.from_profile("preply_entp", role="SNOWFLAKE_ENTP_ANALYST_PII")

You can also select the profile via SNOWFLAKE_CONNECTION_NAME=preply_entp in the environment. When a profile is used, SNOWFLAKE_* env vars and Databricks secrets are NOT consulted, so profiles from other accounts can't be clobbered by stale environment settings.

For local interactive work, SNOWFLAKE_AUTHENTICATOR=externalbrowser is supported. An explicitly requested authenticator always wins: setting externalbrowser uses browser SSO even if a SNOWFLAKE_PRIVATE_KEY/SNOWFLAKE_PRIVATE_KEY_PATH happens to be discoverable in the environment or a secret scope. SSO tokens are cached in the OS keychain, so the browser login only happens once per token lifetime. (Note: externalbrowser works with DataConnector only; SFConnector rejects it, since Spark jobs block on the interactive browser SSO.) For Databricks and Spark jobs, use key-pair auth instead. The connector reads default Databricks personal-scope secrets automatically:

databricks secrets put-secret [e-mail] snowflake_key --bytes-value """$(cat rsa_key.p8)"""
databricks secrets put-secret [e-mail] snowflake_key_pass --string-value """<password>"""

Then build Spark connector options without opening a browser:

from ml_analytics import DataConnector

dc = DataConnector(engine="snowflake", secret_scope="[e-mail]")
options = dc.snowflake_spark_options()

df = (
    spark.read.format("net.snowflake.spark.snowflake")
    .options(**options)
    .option("query", "SELECT 1 AS col_1")
    .load()
)

Query Snowflake With Spark (SFConnector)

On Databricks, SFConnector reads Snowflake directly as Spark DataFrames and can persist results into Unity Catalog tables. It reuses the same SNOWFLAKE_* settings and key-pair secrets as DataConnector, and only imports PySpark when a query/write method runs.

from ml_analytics import SFConnector

sf = SFConnector()  # reads SNOWFLAKE_* env vars; secret scope inferred from SNOWFLAKE_USER

# Spark DataFrame
df = sf.sql("SELECT * FROM cds.dim_tutor LIMIT 1000")

# pandas DataFrame
pdf = sf.sql("SELECT 1 AS col_1", return_pandas=True)

# run a query from a .sql file (relative to project root), with templating
df = sf.sql("queries/experiment.sql", days=14)

# pull and save the result to a Unity Catalog table in one call
sf.sql("queries/experiment.sql", save_table=True, schema="analytics", table="exp")

# or save any Spark DataFrame to Unity Catalog
sf.save_to_uc(df, table="exp", schema="analytics", catalog="prod")

# save a YAML-ordered folder of SQL queries as Unity Catalog tables
df = sf.save_pipeline_to_uc(
    "queries/churn_pipeline",
    pipeline="daily",
    catalog="prod",
    schema="analytics",
)

Credentials resolve per field as: explicit argument → SNOWFLAKE_* environment variable → Databricks secret. See the Snowflake Spark Connector guide for credential setup and all options.

Manage Unity Catalog Tables (SparkTableManager)

When you already have a DataFrame (from any source) and just need Spark / Unity Catalog table operations, use SparkTableManager. It accepts Spark, pandas, or polars DataFrames and imports PySpark lazily, the same way SFConnector does.

from ml_analytics import SparkTableManager

tm = SparkTableManager(catalog="prod", schema="analytics")

# run Spark SQL (Spark DataFrame, or pandas with return_pandas=True)
df = tm.sql("SELECT * FROM prod.analytics.lessons WHERE country = 'US'")

# save a Spark / pandas / polars DataFrame to a Unity Catalog Delta table
tm.save_to_uc(df, table="lessons_us", comment="US lessons")

# convert pandas/polars to Spark (optionally with an explicit schema)
sdf = tm.to_spark(pandas_df, schema="user_id long, country string")

# read back, optimize, or drop
again = tm.read_table("lessons_us")
tm.optimize_uc_table("lessons_us", zorder_by="country")
tm.drop_table("lessons_us")

See the Spark Table Manager guide for all options.

Create A Redshift Table From A DataFrame

dc.create_table_from_dataframe(
    df,
    table="model_scores",
    schema="analytics",
    drop_existing_table=True,
)

Work With S3

from ml_analytics import S3Connector

s3 = S3Connector(bucket="my-analytics-bucket", s3_root="projects/churn")

s3.save_dataframe(df, directory="outputs", file_name="scores")

summary = s3.query(
    """
    SELECT segment, count(*) AS rows
    FROM read_parquet('s3://my-analytics-bucket/projects/churn/outputs/*.parquet')
    GROUP BY segment
    """
)

Read And Write Google Sheets

from ml_analytics import GSheet

gsheet = GSheet(credentials_path="gsheet_credentials.json")

df = gsheet.read_sheet(spreadsheet_id="...", sheet_name="Input")
gsheet.write_sheet(df, spreadsheet_id="...", sheet_name="Results")

OAuth authentication (alternative to a service account)

GSheet can authenticate as your own Google account using OAuth installed-app credentials. Set these env vars and the connector uses OAuth automatically when no service-account credentials are found:

Variable	Required	Description
GOOGLE_OAUTH_CLIENT_ID	yes	OAuth client id (...apps.googleusercontent.com)
GOOGLE_OAUTH_CLIENT_SECRET	yes	OAuth client secret (GOCSPX-...)
GOOGLE_CLOUD_PROJECT	optional	GCP project id (e.g. my-gcp-project)
GSHEET_TOKEN_PATH	optional	Token cache path (default ~/.config/ml-analytics/gsheet_token.json)

The first run opens a browser for one-time consent; the cached refresh token makes later runs non-interactive. Under OAuth, get_service_account_email() returns None.

Log To MLflow

from ml_analytics import ModelManager

manager = ModelManager(model_name="churn-model", user="[e-mail]")

manager.start_run("training")
manager.log_metric("auc", 0.91)
manager.end_run()

For Databricks Unity Catalog model registry, use a three-level model name. When the tracking URI is Databricks, ModelManager automatically sets the MLflow registry URI to databricks-uc for names like catalog.schema.model. The legacy hive_metastore catalog is not treated as a Unity Catalog model registry target; use registry_uri="databricks" and a one-level model name for the workspace model registry.

manager = ModelManager(
    model_name="dev.ml_models.churn_model",
    tracking_uri="databricks",
    user="[e-mail]",
)

Send A Slack Message

from ml_analytics import SlackConnector

slack = SlackConnector()
slack.send_message(channel="#ml-alerts", text="Training finished")

Detailed Guides

Guide	Use It For
AWS Authentication	AWS SSO setup and Python helpers
CLI Commands	Available console commands
Snowflake Spark Connector	SFConnector credential setup, reads, and writes on Spark/Databricks
Spark Table Manager	SparkTableManager — save DataFrames to Unity Catalog, convert pandas/polars, table maintenance
Google Sheets	Sheets setup, sharing, exports, and examples
Slack	Slack token setup and message/file examples
Tunnel Manager	SSH tunnel configuration and CLI usage

Development

Run the standard checks before opening a PR:

uv run ruff check
uv run pytest

CI runs Ruff and pytest on Python 3.11 and 3.12.

Releases

This repository uses Release Please. Conventional commits on main create or update a release PR with the next version and changelog. When that PR is merged, the release workflow builds the package and publishes it to PyPI through Trusted Publishing using the pypi GitHub environment.

Contributing

Keep changes small, covered by tests when behavior changes, and free of environment-specific defaults. Prefer explicit configuration over hidden infrastructure assumptions.