sqlframe

You can enable SQLFrame for BigQuery using one of two patterns:

1. Direct Import: Replace pyspark.sql imports with sqlframe.bigquery. This is recommended when converting an existing PySpark pipeline. Note that many classes will have a BigQuery prefix (e.g., BigQueryDataFrame instead of DataFrame).

2. Activation: Use the activate("bigquery", ...) function. This allows you to continue using pyspark.sql imports while SQLFrame runs the operations on BigQuery behind the scenes. Once activated, pyspark.sql.SparkSession will return a SQLFrame BigQuerySession object.

# Pattern 1: Direct Import
from sqlframe.bigquery import BigQuerySession
from sqlframe.bigquery import functions as F
from sqlframe.bigquery import BigQueryDataFrame

# Pattern 2: Activation
from sqlframe import activate
activate("bigquery", config={"default_dataset": "sqlframe.db1"})
from pyspark.sql import SparkSession

SQLFrame extends BigQuery capabilities by providing a Table class that supports Data Manipulation Language (DML) operations not natively available in PySpark, such as update, delete, and merge.

To obtain a Table object, use the session.table("table_name") method. This returns a BigqueryTable instance. Most DML methods return a LazyExpression object, which must be explicitly executed using the .execute() method to perform the operation on the BigQuery backend.

import google.auth
from google.cloud.bigquery.dbapi import connect
from sqlframe.bigquery import BigQuerySession

# ... setup connection and session ...
session = BigQuerySession(conn=conn, default_dataset="sqlframe.db1")

# Create a DataFrame and save it as a table
session.createDataFrame(data).write.mode("overwrite").saveAsTable("employee")

# Access the table for DML operations
table_employee = session.table("employee")

SQLFrame provides a Table class for Databricks that extends standard PySpark functionality by supporting Data Manipulation Language (DML) operations like update, delete, and merge. You can obtain a Table object (specifically a DatabricksTable) by calling session.table("table_name").

Note that these methods return a LazyExpression object, which must be executed using the .execute() method to perform the operation on the server.

from databricks.sql import connect
from sqlframe.databricks import DatabricksSession

conn = connect(
    server_hostname="dbc-xxxxxxxx-xxxx.cloud.databricks.com",
    http_path="/sql/1.0/warehouses/xxxxxxxxxxxxxxxx",
    access_token="YOUR_TOKEN",
    auth_type="access_token",
    catalog="catalog",
    schema="schema",
)
session = DatabricksSession(conn=conn)

# Returns a DatabricksTable object
table_employee = session.table("employee")

You can enable SQLFrame for Snowflake using two different approaches depending on your workflow:

1. Direct Import: Best for converting existing PySpark pipelines. Replace pyspark.sql imports with sqlframe.snowflake. Note that many classes will have a Snowflake prefix (e.g., SnowflakeDataFrame instead of DataFrame).

2. Activation: Best if you want to continue using the standard pyspark.sql API. By calling activate("snowflake", conn=conn), any subsequent calls to pyspark.sql.SparkSession will return a SQLFrame SnowflakeSession object, and all operations will be executed directly on Snowflake.

# Direct Import approach
from sqlframe.snowflake import SnowflakeSession
from sqlframe.snowflake import functions as F
from sqlframe.snowflake import SnowflakeDataFrame

# Activation approach
from sqlframe import activate
from snowflake.connector import connect
import os

conn = connect(
    account=os.environ["SNOWFLAKE_ACCOUNT"],
    user=os.environ["SNOWFLAKE_USER"],
    password=os.environ["SNOWFLAKE_PASSWORD"],
    warehouse=os.environ["SNOWFLAKE_WAREHOUSE"],
    database=os.environ["SNOWFLAKE_DATABASE"],
    schema=os.environ["SNOWFLAKE_SCHEMA"],
)
activate("snowflake", conn=conn)

from pyspark.sql import SparkSession
spark = SparkSession.builder.getOrCreate()

SQLFrame can be enabled for Redshift in two ways:

1. Direct Import: Replace pyspark.sql imports with sqlframe.redshift imports. Note that many classes will have a Redshift prefix (e.g., RedshiftDataFrame instead of DataFrame).
2. Activation: Use the activate("redshift", conn=conn) function. This allows you to continue using the standard pyspark.sql API while SQLFrame runs the operations directly on Redshift behind the scenes.

Both methods require a redshift_connector.Connection object.

# Method 1: Direct Import
from sqlframe.redshift import RedshiftSession
from sqlframe.redshift import functions as F
from sqlframe.redshift import RedshiftDataFrame

# Method 2: Activation
import os
from redshift_connector import connect
from sqlframe import activate

conn = connect(
    user="user",
    password=os.environ["PASSWORD"],
    database="database",
    host="xxxxx.xxxxxx.region.redshift-serverless.amazonaws.com",
    port=5439,
)
activate("redshift", conn=conn)

from pyspark.sql import SparkSession
# SparkSession is now a RedshiftSession