API reference

Spark runtimeDownload PDF (3.0 MB)

This section documents the Groundzero helper API that is available in the Spark runtime. Standard PySpark (DataFrame, functions, types, MLlib) is documented by Apache Spark and is not repeated here.

Conventions#

  • Import the session factory once: from GZ.SparkUtils import sparkSession.
  • The object it returns (called spark in this guide) is the Groundzero session. Every connector method is a method on it.
  • Connections are always referenced by connName, the connection name in Groundzero. Credentials never appear in code.
  • Read and execute methods return a Spark DataFrame. Write methods take the DataFrame as the first argument (df) and generally return None.
  • Method and parameter names use camelCase and are case sensitive.
  • Argument names in the signatures below are the real parameter names; pass them by keyword for readability.

Parameters not described here

Parameters that a method accepts but that this guide does not describe are passed straight through to the underlying connector.

Session#

sparkSession#

python
sparkSession(appName, logLevel="WARN", siteName=None, siteNames=None, **kwargs)

Creates (or returns) the Groundzero session for the current run.

Parameter Type Default Description
appName str required Application name shown in Spark and the logs
logLevel str "WARN" Spark log level, such as INFO or ERROR
siteName str None Site used for every REST-backed lakehouse connection
siteNames dict None Mapping of connection name to site
**kwargs any none Extra Spark settings, for example **{"spark.x": "y"}

siteNames also accepts a list of site names (a single entry applies to every REST-backed connection; a list of the same length maps by connection order) or a list of dictionaries with connName (or connectionName, connection, name) and siteName (or site).

Returns. The Groundzero session object.

Raises. ValueError if you try to move a lakehouse connection to a different site in the same session: GZLAKEHOUSE connection 'x' is already configured for site 'a'; cannot reconfigure it to 'b' in the same Spark session.

Behavior:

  • Calling sparkSession again in the same notebook or reused session returns the existing session. Extra Spark settings in **kwargs take effect only on the first call.
  • Your settings are applied on top of the Groundzero defaults (see Tune performance).
  • Lakehouse catalogs are registered for every GZLAKEHOUSE connection in the run.
python
spark = sparkSession(
    "DailyOrders",
    "INFO",
    siteNames={"acme-lakehouse": "acme", "partner-lakehouse": "partner"},
)

Session object#

The session is a restricted wrapper around a Spark session.

Member Behavior
spark.sql(sqlQuery, *args, **kwargs) Runs Spark SQL; catalog and policy checks apply
spark.createDataFrame(...) Passed through to Spark
spark.table(...), spark.range(...) Passed through to Spark
spark.conf, spark.catalog, spark.udf Passed through to Spark
spark.read, spark.write Blocked; use the connector methods
spark.sparkContext Blocked

Accessing a blocked member raises AttributeError: 'sparkSession' object has no attribute 'read'. Use the connector's read* method instead. On a DataFrame, df.write works as in Spark, but connectors (which attach credentials, authorization, audit and lineage) are the supported route.

Lakehouse (Iceberg) methods#

executeQuery#

python
spark.executeQuery(connName, sqlQuery, siteName=None)

Runs SQL against a GZLAKEHOUSE connection and returns the result.

Parameter Type Default Description
connName str required GZLAKEHOUSE connection name
sqlQuery str required SQL; reference tables as database.table
siteName str None Site to route to; falls back to the session site

Returns. A Spark DataFrame.

Raises.

  • PermissionError: authentication failed, the user is not authorized for a referenced table, or the connection is external (read-only) and the statement is not SELECT, SHOW, DESCRIBE or EXPLAIN.
  • ConnectionError: the lakehouse endpoint could not be reached.
  • RuntimeError: the endpoint returned a non-success status, or routed to a warehouse that does not match the connection.
  • ValueError: the site is unknown or the connection has no warehouse name.

Every call is recorded in the lakehouse audit log (success or failure, with the duration), and lineage is reported for the tables it touches.

python
df = spark.executeQuery(
    connName="acme-lakehouse",
    sqlQuery="SELECT id, status FROM sales.orders WHERE status = 'open'",
    siteName="acme",
)

writeGzLakehouse#

python
spark.writeGzLakehouse(
    df, connName, tableName, database=None, mode="overwrite",
    partitionBy=None, options=None, siteName=None, streaming=False,
    checkpointLocation=None, outputMode="append", triggerTime=None,
    trigger=None,
)

Writes a DataFrame to a lakehouse table.

Parameter Type Default Description
df DataFrame required Data to write; streaming frames start a stream
connName str required GZLAKEHOUSE connection name
tableName str required Target table
database str None Target database; defaults to the connection's
mode str "overwrite" append or overwrite (endpoint connections)
partitionBy list None Partition columns; batch writes on legacy connections only
options dict None Extra writer options, such as fanout-enabled for streams
siteName str None Site to route to
streaming bool False Start a Structured Streaming write
checkpointLocation str None Required for streaming writes
outputMode str "append" append or complete for streams
triggerTime str None Processing-time trigger such as "1 minute"
trigger dict None Trigger arguments such as {"availableNow": True}

Returns. For a batch write to a lakehouse endpoint, a dict with status, table, warehouse, catalog, site, mode and via. For a stream, the running streaming query.

Raises.

  • ValueError: mode is not append or overwrite; database is missing; the site is missing; partitionBy is used with a stream.
  • PermissionError: the user may not write the table, or the connection is external.
  • ConnectionError, RuntimeError: the lakehouse endpoint is unreachable or rejected the commit.

A DataFrame whose isStreaming flag is true is written as a stream even when streaming is not set. The write is authorized per table, audited, and its schema and lineage are captured.

getLakehouseCatalog#

python
spark.getLakehouseCatalog(connName, siteName)

Returns the Spark catalog alias for a REST-backed connection and site, so you can write fully qualified SQL for spark.sql.

Parameter Type Default Description
connName str required GZLAKEHOUSE connection name
siteName str required Site that fixes the catalog

Returns. str, the catalog alias. Raises. ValueError if the connection has no lakehouse endpoint.

python
catalog = spark.getLakehouseCatalog("acme-lakehouse", "acme")
spark.sql(f"SELECT COUNT(*) FROM {catalog}.sales.orders").show()

setLakehouseSite#

python
spark.setLakehouseSite(connName, siteName)

Registers the catalog for a site ahead of use. It does not change the active catalog. Returns. None. Raises. ValueError if the connection has no warehouse name or no lakehouse endpoint.

verifyLakehouseConnection#

python
spark.verifyLakehouseConnection(connName, siteName=None)

Probes the lakehouse endpoint and checks that it routes to the connection's warehouse. The probe is anonymous: it checks reachability and routing, not authorization.

Returns. dict with the keys provider_url, warehouse, site_used and prefix. Raises. ConnectionError, RuntimeError or ValueError with a descriptive message.

Bulk loads into the lakehouse#

python
spark.bulkLoadGzLakehouseFromSnowflake(
    connName, query, lakehouseConn, tableName, database=None,
    mode="overwrite", warehouseSize=None, storageIntegration=None,
    stagingLocation=None, maxFileSize=DEFAULT_MAX_FILE_SIZE_BYTES,
    siteName=None,
)
spark.bulkLoadGzLakehouseFromRedshift(
    connName, query, lakehouseConn, tableName, database=None,
    mode="overwrite", iamRole=None, stagingLocation=None,
    maxFileSize=DEFAULT_MAX_FILE_SIZE_BYTES, siteName=None,
)

The warehouse unloads the query result as Parquet under the target table's location, and the files are registered with one Iceberg commit. No row passes through Spark. The default for maxFileSize is 256 MiB.

Parameter Type Default Description
connName str required Source Snowflake or Redshift connection
query str required Query whose rows are loaded
lakehouseConn str required GZLAKEHOUSE connection of the target
tableName str required Target table; created from the file schema if missing
database str None Target database; defaults to the connection's
mode str "overwrite" overwrite replaces the rows, append adds
warehouseSize str None Snowflake only; warehouse size during the unload
storageIntegration str None Snowflake only; storage integration to write with
iamRole str None Redshift only; role to unload with
stagingLocation str None Prefix inside the table location for the files
maxFileSize int 256 MiB Upper bound in bytes per unloaded file
siteName str None Site to route to

Returns. A summary of the rows and files registered, or a summary with via: "spark_write" when the files had to be written with Spark. Raises. ValueError if the Redshift method is called with a connection that is not a Redshift connection.

Object storage and files#

readS3#

python
spark.readS3(connName, fileName, format="parquet", schema=None, options=None)
Parameter Type Default Description
connName str required S3 connection name
fileName str required Key or prefix inside the connection's bucket
format str "parquet" Any Spark format: csv, json, parquet, orc
schema schema None Spark schema; skips inference
options dict None Reader options, for example {"header": "true"}

Returns. A DataFrame. Raises. Exception("Error reading file from S3: ...") wrapping the underlying cause.

writeS3#

python
spark.writeS3(df, connName, filePath, format, mode, options=None, coalesce=False)
Parameter Type Default Description
df DataFrame required Data to write
connName str required S3 connection name
filePath str required Target key or prefix
format str required Output format
mode str required overwrite, append, ignore or errorifexists
options dict None Writer options
coalesce bool False Write a single file (df.coalesce(1))

Returns. None.

Coalesce funnels all data through one task

coalesce=True makes one executor write the entire result. Use it only for small outputs.

writeStreamToS3#

python
spark.writeStreamToS3(df, connName, filePath, format, outputMode,
                      checkpointLocation, coalesce=False, options=None)

Writes each micro-batch of a streaming DataFrame to S3 in overwrite mode. Returns. The running streaming query.

Relational databases and warehouses#

readJDBC#

python
spark.readJDBC(connName, query, options=None)

Reads the result of query from any JDBC connection (PostgreSQL, MySQL, SkySQL, Oracle, SQL Server, Redshift, Teradata, IBM Db2, SAP and SAP HANA, Synapse, CockroachDB, TimescaleDB, ClickHouse, Databricks, Trino, Presto, Vertica, Dremio).

Parameter Type Default Description
connName str required JDBC connection name
query str required SQL sent to the database
options dict None Extra Spark JDBC options, such as fetchsize

Returns. A DataFrame. Raises. ValueError("Connection x not found or is not a JDBC type."). TLS options for Redshift are enforced by the runtime.

writeJDBC#

python
spark.writeJDBC(df, connName, tableName, mode, options=None)
Parameter Type Default Description
df DataFrame required Data to write
connName str required JDBC connection name
tableName str required Target table
mode str required append, overwrite, ignore or errorifexists
options dict None Extra Spark JDBC options

Raises. ValueError for an unknown connection, and TypeError("Use writeSnowflake() for connection x.") for Snowflake connections.

executeJDBC#

python
spark.executeJDBC(connName, query)

Runs a DDL or DML statement through the JDBC driver and commits it. Prints DDL statement executed successfully! and returns None. Errors are printed and re-raised.

Snowflake#

python
spark.executeSnowflake(connName, query, options=None)
spark.writeSnowflake(df, connName, tableName, mode="append", extraOptions=None)
spark.executeSnowflakeStatement(connName, statement)
Method Returns
executeSnowflake A DataFrame with the query result
writeSnowflake None; repartitions to at least 200 partitions first
executeSnowflakeStatement The string Snowflake statement executed successfully.

options and extraOptions are merged into the connector options. Use executeSnowflakeStatement for DDL and session statements.

LLM methods#

complete#

python
spark.complete(connName, df, model=None, promptColumn="prompt",
               promptTemplate=None, systemPrompt=None, systemPromptColumn=None,
               outputColumn="llm_response", temperature=0.0, maxTokens=None,
               responseFormat=None, continueOnError=False, timeout=120,
               **providerOptions)

Adds a completion to every row of a DataFrame using the provider of the connection (OpenAI, Azure OpenAI, Anthropic, AWS Bedrock, Cohere, Gemini, Fireworks AI, Together AI, Hugging Face and xAI). The calls run inside the executors, so the work is distributed and no rows are collected to the driver.

Parameter Type Default Description
connName str required LLM connection name
df DataFrame required Input rows
model str None Model; for Azure the deployment name; for Bedrock the model ID
promptColumn str "prompt" Column holding the prompt text
promptTemplate str None Template with {{column}} or ${column} fields
systemPrompt str None System prompt applied to every row
systemPromptColumn str None Column with a per-row system prompt
outputColumn str "llm_response" Name of the response column
temperature float 0.0 Sampling temperature
maxTokens int None Response limit
responseFormat str None Provider response format, for example json
continueOnError bool False Record per-row errors instead of failing
timeout int 120 Request timeout in seconds (not used by Bedrock)

providerOptions accepts provider-only settings, such as useConverse for Bedrock or returnFullText for Hugging Face.

Returns. The input DataFrame plus these columns, where <out> is outputColumn: <out>, <out>_status (success or error), <out>_error, <out>_provider, <out>_model, <out>_prompt_tokens, <out>_completion_tokens, <out>_total_tokens and <out>_latency_ms.

Raises. ValueError if the connection is unknown (Connection 'x' was not found.), has an unsupported type (Connection 'x' has unsupported LLM type 'y'. Supported types: ...), a referenced column is missing (Missing columns: [...]. Available: [...]), or a Bedrock call has no model (AWS Bedrock model must be set in the connection or call.). With continueOnError=False the first failing row fails the job.

Requests are retried with exponential backoff on HTTP 429 and 5xx responses.

python
tickets = spark.createDataFrame(
    [(1, "Printer jams on page two"), (2, "Cannot reset my password")],
    ["id", "text"],
)
labeled = spark.complete(
    connName="acme-openai",
    df=tickets,
    promptTemplate="Label the ticket as hardware or account: {{text}}",
    outputColumn="label",
    maxTokens=8,
    continueOnError=True,
)

Fine-tuning and batch operations#

Provider-neutral entry points route by connection type:

python
spark.fineTune(connName, df, **kwargs)
spark.distill(connName, df, **kwargs)
spark.listFineTuneJobs(connName, **kwargs)
spark.getFineTuneJob(connName, **kwargs)
spark.cancelFineTuneJob(connName, **kwargs)
spark.listModels(connName, **kwargs)

Each forwards its arguments to the provider method named by the operation plus the provider suffix, for example fineTuneOpenAI, fineTuneAwsBedrock or listModelsTogetherAI. For AWS Bedrock, cancel maps to stopFineTuneJobAwsBedrock. If the provider does not implement the operation, a NotImplementedError is raised: fineTune is not available for connection type 'x'.

Provider suffix Typical methods
OpenAI, AzureOpenAI fineTune, distill, jobs, batch jobs, deployments
AwsBedrock fineTune, distill, jobs (stop), batch jobs, deployments
TogetherAI, FireworksAI fineTune, jobs, batch jobs, deployments
Cohere, Gemini, HuggingFace, Anthropic complete, models, batch jobs, deployments
Unsloth fineTuneUnsloth, listModelsUnsloth (GPU runtime)
XAI completeXAI, listModelsXAI

Direct fine-tuning is unavailable for Anthropic, Cohere, the Gemini API and the generic Hugging Face connector. Use AWS Bedrock for eligible Claude models, Vertex AI for Gemini tuning, and Unsloth or AutoTrain for Hugging Face models. OpenAI and Azure OpenAI require at least 10 training examples.

The batch method families follow the same naming: createBatchJob<Provider>, listBatchJobs<Provider>, getBatchJob<Provider>, cancelBatchJob<Provider> and getBatchResults<Provider>. Deployment methods follow createDeployment<Provider>, listDeployments<Provider>, getDeployment<Provider> and deleteDeployment<Provider>. Which ones exist for a provider is listed in the table above; the method listing for each connector is in the connector reference below.

Data quality#

python
spark.createDataQualityChecker(df, connectionName=None, config=None,
                               references=None)
spark.runDataQualityChecks(df, connectionName=None, config=None,
                           includeLlm=True, references=None)
spark.runTableDataQualityChecks(tableName, connectionName=None, config=None,
                                includeLlm=True, references=None)
spark.runQueryDataQualityChecks(query, connectionName=None, config=None,
                                includeLlm=True, references=None)
spark.saveDataQualityReport(report, path, reportFormat="json")
Parameter Type Default Description
df DataFrame required Data to check (streaming frames are not allowed for local work)
tableName str required Catalog table to load and check
query str required Spark SQL whose result is checked
connectionName str None LLM connection used for the narrative
config QualipilotConfig None Checks to run
includeLlm bool True Ask the LLM to narrate the results
references dict None Reference data forwarded to Qualipilot
report QualityReport required Report returned by a run method
path str required Output path
reportFormat str "json" Output format, for example html

Returns. runDataQualityChecks, runTableDataQualityChecks and runQueryDataQualityChecks return a QualityReport with the check results and llm_report. createDataQualityChecker returns a checker you can use as a context manager and call run(include_llm=...) on.

Only a compact aggregate of the deterministic results goes to the LLM. Row samples and top values are excluded.

Local record linkage#

spark.runLocalRecordLinkage(data, config, rightData=None, *, maxRows)
spark.runLocalRecordDeduplication(data, config, consolidation, *, maxRows)
spark.consolidateLocalRecords(data, idColumn, clusters, config, *, maxRows)

Run Qualipilot's native linkage on small datasets, materialized directly as driver-local Polars DataFrames. maxRows is mandatory and must be positive; the runtime applies limit(maxRows + 1) and fails if the data exceeds it.

Raises. ValueError with messages such as maxRows must be a positive integer, data exceeds maxRows=N, rightData is required in link mode, rightData is only valid in link mode, and driver-local Qualipilot operations reject streaming data. Nested, variant, duplicate or case-conflicting columns fail before collection. maxRows limits rows, not width or candidate pairs, so select only the columns you need.

PDF extraction#

parsePDF#

python
spark.parsePDF(s3ConnName=None, inputUri=None, outputPrefix=None,
               outputFormat="csv", outputFormats=None, inputFile=None,
               connName=None, hybrid=False, ocrMode="auto",
               ocrStack="ppocrv6", strict=False, password=None,
               language="eng", maxPages=None, maxFileBytes=None,
               llmProvider=None, llmConnName=None, llmModel=None,
               llmBaseUrl=None, llmApiKey=None, outputS3ConnName=None,
               returnAgentPages=False, contentAddressed=False,
               inlineAgentPages=True, pageRange=None)

Parses one PDF stored in S3 and writes the result back to S3. The llm* parameters configure the LLM-assisted hybrid mode and are used together with hybrid=True.

Parameter Type Default Description
s3ConnName str None S3 connection of the source (or use connName)
inputUri str None PDF location; inputFile is an alias
outputPrefix str generated Output prefix; defaults to pdf-extractor/<uuid>
outputFormat str "csv" json, md, txt, csv or xlsx
outputFormats list None Several formats at once
ocrMode str "auto" off, auto or all
ocrStack str "ppocrv6" classic or ppocrv6
language str "eng" OCR language
strict bool False Fail instead of returning partial results
password str None Password for encrypted PDFs
maxPages int None Page limit
maxFileBytes int None File size limit
outputS3ConnName str None S3 connection for output; defaults to the source

Returns. A dict with status, input_uri, output_prefix, output_format, result_uri, artifacts (including a manifest), page_count and sourceSha256.

Raises. ValueError("s3ConnName is required") and ValueError("inputUri or inputFile is required"); ValueError("pageRange requires returnAgentPages=True").

readPDF and readPDFResult#

python
spark.readPDF(s3ConnName, inputUri=None, outputPrefix=None, outputFormat="csv",
              outputFormats=None, readOptions=None, outputS3ConnName=None,
              **options)
spark.readPDFResult(resultUri, outputFormat="csv", s3ConnName=None,
                    readOptions=None)

readPDF parses a PDF and returns the result as a DataFrame. readPDFResult reads an existing result. Spark reads csv (without a header by default), json and the text formats txt, text, md, markdown. Other formats raise ValueError("Spark cannot read 'x' directly").

parsePDFCorpus#

python
spark.parsePDFCorpus(s3ConnName, sourcePrefix=None, inputUris=None,
                     outputPrefix=None, outputS3ConnName=None,
                     pageRangeSize=64, partitions=None, ocrMode="auto",
                     language="eng", strict=False, password=None,
                     maxFileBytes=None, maxDocuments=10_000)

Parses an S3 prefix, or an explicit list, as restartable distributed work. Discovery is recursive and finds .pdf files regardless of case. Results are content-addressed: re-running with the same source and output prefix skips page ranges that were already committed.

Parameter Type Default Description
s3ConnName str required S3 connection of the source
sourcePrefix str None Folder to scan (give this or inputUris)
inputUris list None Explicit PDF list
outputPrefix str None Where results go
pageRangeSize int 64 Pages per unit of work
partitions int None Parallel parsers, capped (default cap 2)
maxDocuments int 10_000 Document limit

Returns. A dict including status, documentCount and manifestUri.

The cap on partitions exists because every parser worker loads a complete document-model set; size executor memory before raising it.

Agents#

listAgents#

python
spark.listAgents(connName)

Returns a list[dict] of the Agents visible to the connection's service account.

invokeAgent#

spark.invokeAgent(connName, agentId, message, sessionId=None, *,
                  requestId=None, expectedAgentVersion=None,
                  disableTools=False, pdfSourceConnName=None,
                  pdfSourceObject=None, pdfSinkConnName=None,
                  pdfSinkPrefix=None)

Submits the message in the background and waits for the final result, polling for up to 15 minutes so no HTTP request stays open for the whole run.

Parameter Type Default Description
connName str required Agent connection
agentId str required Agent identifier (UUID)
message str required Non-empty message
sessionId str None UUID of an existing conversation
requestId str None Idempotency identifier
expectedAgentVersion int None Fail if the live version differs; must be >= 1
disableTools bool False Run without tools; not allowed with PDFs
pdfSourceConnName str None S3 connection of a PDF to bind
pdfSourceObject str None PDF location
pdfSinkConnName str None S3 connection for extracted pages
pdfSinkPrefix str None Prefix for extracted pages

Returns. The final invocation result as a dict, including pdfArtifacts when a PDF was bound.

Raises. ValueError (message must be a non-empty string, expectedAgentVersion must be a positive integer, disableTools must be a boolean, disableTools cannot be used with PDF tool bindings), and TimeoutError with the run ID if polling times out. A timeout does not cancel or resubmit the run; inspect the existing run before starting another.

invokeAgentPDFCorpus#

spark.invokeAgentPDFCorpus(connName, agentId, message, pdfSourceConnName,
                           pdfSinkPrefix, *, pdfSourcePrefix=None,
                           pdfSourceObjects=None, pdfSinkConnName=None,
                           pageRangeSize=64, partitions=None,
                           summaryWorkers=4, summaryInputChars=100_000,
                           maxDocuments=10_000, maxAgentCalls=50_000,
                           ocrMode="auto", language="eng", strict=False,
                           password=None, maxFileBytes=None)

Runs an approved Agent over every complete PDF under a prefix. Pass exactly one of pdfSourcePrefix or pdfSourceObjects.

Parameter Type Default Description
connName str required Agent connection
agentId str required Agent identifier
message str required Instruction applied to every document
pdfSourceConnName str required S3 connection of the documents
pdfSinkPrefix str required Prefix for results
summaryWorkers int 4 Concurrent document workflows, 1 to 32
maxDocuments int 10_000 Document limit
maxAgentCalls int 50_000 Shared Agent-call budget

Returns. A dict with status, outputMode, resultsUri, manifestUri, processedDocumentCount and skippedDocumentCount. Each row at resultsUri links to a JSON object with the validated data. The older names summaryResultsUri, summaryManifestUri, summaryUri and summarizedDocumentCount remain aliases.

The corpus resolves the approved Agent version and response schema before any PDF work. With a JSON response schema, the output mode is structured; without one, the Agent returns cited facts. Unchanged results are reused on later runs. Documents that fail are recorded in the run index and do not stop the others.

Jobs, mail and clients#

executeJob#

python
from GZ.JobUtils import executeJob

executeJob(jobNames)

Runs the named sub-jobs of the current job in sequence, each under the same restrictions as a normal job.

Parameter Type Default Description
jobNames list[str] required Names of sub-jobs to run, in order

Returns. dict mapping each job name to its output. Raises. ValueError when none of the names match a sub-job, and any exception from a sub-job (the remaining jobs do not run).

sendEmail#

python
from GZ.SendEmail import sendEmail

sendEmail(subject, body, to_email, from_email, **kwargs)

Sends an email through the Groundzero notification service. Extra keyword arguments are added to the message.

Parameter Type Default Description
subject str required Subject line
body str required Message body
to_email str required Recipient address
from_email str required Sender address

Returns. None. Raises. Any error from the notification service.

getClient#

python
from GZ.Client import getClient

client = getClient()

Returns a client factory. Its methods build authenticated native clients from a connection name; each takes connName (the SQL engine method takes conn_name).

Method Returns
getSQLEngine(conn_name) SQLAlchemy engine (PostgreSQL, MySQL, Oracle, Db2, SQL Server, SAP, Redshift, Exasol, CockroachDB)
getS3Client(connName) boto3 S3 client
getS3FSClient(connName) s3fs file system
getSnowflakeClient(connName) Snowflake connection
getMongoClient(connName) MongoDB client
getAzureBlobClient(connName) Azure Blob client
getKinesisClient, getDynamoDBClient, getAthenaClient AWS clients
getBigQueryClient, getGCSClient, getGoogleDriveClient Google clients
getCassandraSession, getRedshiftClient, getElasticClient Database clients
getOpenAIClient, getAzureOpenAIClient, getAwsBedrockClient LLM clients
getCohereClient, getTogetherAIClient, getHuggingFaceClient LLM clients
getFireworksAIClient, getXAIClient, getMcpClient LLM and MCP clients

Further methods exist for Salesforce, SFTP, CosmosDB, Exasol, Timescale, Pinecone, Firebolt, Teradata, Neo4j, InfluxDB, DuckDB, Business Central, Shopify, Marketo, Couchbase, ClickHouse, OneDrive, Splunk, Databricks SQL, Synapse, Event Hubs, SharePoint, ScyllaDB and Weaviate, plus read_rest and write_rest helpers for REST connections.

Placeholders and outputs#

Placeholders#

Placeholder Replaced with
{STRTTM} Run start time, YYYY-MM-DD HH:MM:SS
{ENDTM} Timestamp taken just before your script starts; in notifications, the end time
{CLNTNM} Client name
{SITENM} Site name
{PRJNM} Project name
{JOBNM} Job name
{STS} Status
{ERR} Error message
{RTE} Runtime name
{RGN} Region
{USR} User name of the caller
{USRGRPS} Groups of the caller

{STS} and {ERR} are meant for notification messages. In job code they are empty. In code, each placeholder becomes a Python string literal that already includes single quotes (for example "'2026-10-01 09:30:00'"). {USRGRPS} expands to a comma-separated list of such literals. A name that is not in the table is left unchanged, so ordinary braces in strings and dictionaries are safe unless they match a name exactly. The ${NAME} spelling is equivalent.

outputDF#

After your script ends, the runtime reads the global outputDF:

Value Handling
Spark DataFrame Up to 100,000 rows are collected; schema from the DataFrame
pandas DataFrame Converted with to_dict(orient="records")
Any other iterable Converted with list(...)
Not defined No preview; the job still succeeds

Connector method index#

Every connector below is available as methods on the session. df is a Spark DataFrame; connName is the Groundzero connection name; mode is a Spark save mode unless noted. Methods ending in Arrow exchange data through Apache Arrow tables.

Look up syntax

The Groundzero assistant can look up the Spark syntax of any connector method, so you do not need to memorize this list.

Object storage and files#

# S3
spark.readS3(connName, fileName, format='parquet', schema=None, options=None)
spark.writeS3(df, connName, filePath, format, mode, options=None, coalesce=False)
spark.writeStreamToS3(df, connName, filePath, format, outputMode, checkpointLocation,
    coalesce=False, options=None)
# GCS
spark.getCredentialsFileGCS(credentialsFile)
spark.readGcs(connName, fileName, format, options)
spark.writeGcs(df, connName, fileName, format, mode, options=None, coalesce=False)
# AZUREBLOB
spark.readAzureBlob(connName, filePath, options=None)
spark.writeAzureBlob(df, connName, filePath, format, mode, options=None)
# SFTP
spark.readSFTP(connName, fileType, filePath)
spark.writeSFTP(df, connName, fileType, filePath, mode='overwrite')
spark.listSFTP(connName, remotePath='.')
spark.deleteSFTP(connName, remotePath)
spark.readSFTPArrow(connName, fileType, filePath, **options)
spark.writeSFTPArrow(df, connName, fileType, filePath, mode='overwrite', **options)
spark.listSFTPArrow(connName, remotePath='.', recursive=False)
spark.deleteSFTPArrow(connName, remotePath, recursive=False)
spark.mkdirSFTPArrow(connName, remotePath)
spark.renameSFTPArrow(connName, srcPath, dstPath)
spark.existsSFTPArrow(connName, remotePath)
# GOOGLEDRIVE
spark.readGoogleDriveFiles(connName, folderId=None, query=None, limit=None)
spark.readGoogleDriveFileContent(connName, fileId)
spark.writeGoogleDriveFiles(df, connName, parentFolderId=None, batchSize=10,
    maxPartitions=None)
spark.deleteGoogleDriveFile(connName, fileId)
spark.createGoogleDriveFolder(connName, folderName, parentFolderId=None)
spark.readGoogleDriveFilesArrow(connName, folderId=None, query=None, limit=None,
    columns=None, options=None)
# ONEDRIVE
spark.readOneDriveFiles(connName, folderPath=None, fileExtensions=None, limit=None,
    batchSize=1000, userPrincipalName=None, driveId=None)
spark.writeOneDriveFiles(df, connName, folderPath=None, batchSize=10,
    maxPartitions=None, userPrincipalName=None, driveId=None)
spark.deleteOneDriveFile(connName, fileId=None, filePath=None, userPrincipalName=None,
    driveId=None)
spark.readOneDriveFilesArrow(connName, folderPath=None, limit=None, batchSize=100,
    userPrincipalName=None, driveId=None, columns=None, options=None)
# SHAREPOINT
spark.readSharePoint(connName, listName, select=None, filter=None, orderBy=None,
    expand=None, top=None, maxResults=5000)
spark.writeSharePoint(df, connName, listName, operation='create', batchSize=100)
spark.readSharePointArrow(connName, listName, select=None, filter=None, orderBy=None,
    expand=None, top=None, maxResults=None, columns=None, options=None)
spark.writeSharePointArrow(df, connName, listName, operation='create', batchSize=100,
    options=None)

Table formats#

# DELTA
spark.readDelta(connName, path=None, tableName=None, options=None, **cloudOptions)
spark.writeDelta(df, connName, path=None, tableName=None, mode='append', options=None,
    **cloudOptions)
spark.executeDelta(query)
# HUDI
spark.readHudi(connName, tableName, options=None)
spark.writeHudi(df, connName, tableName, writeOperation='upsert', options=None)

Relational databases and warehouses#

# JDBC
spark.readJDBC(connName, query, options=None)
spark.writeJDBC(df, connName, tableName, mode, options=None)
spark.executeJDBC(connName, query)
spark.bulkLoadGzLakehouseFromRedshift(connName, query, lakehouseConn, tableName,
    database=None, mode='overwrite', iamRole=None, stagingLocation=None,
    maxFileSize=DEFAULT_MAX_FILE_SIZE_BYTES, siteName=None)
# SNOWFLAKE
spark.executeSnowflakeStatement(connName, statement)
spark.executeSnowflake(connName, query, options=None)
spark.writeSnowflake(df, connName, tableName, mode='append', extraOptions=None)
spark.bulkLoadGzLakehouseFromSnowflake(connName, query, lakehouseConn, tableName,
    database=None, mode='overwrite', warehouseSize=None, storageIntegration=None,
    stagingLocation=None, maxFileSize=DEFAULT_MAX_FILE_SIZE_BYTES, siteName=None)
# GOOGLEBIGQUERY
spark.getCredentialsFileBQ(credentialsFile)
spark.executeBigQuery(connName, tableName, query=None)
spark.writeBigQuery(df, connName, tableName, writeMethod, mode)
# ATHENA
spark.readAthena(connName, region=None, query=None, options=None, workGroup=None,
    catalog=None, sessionToken=None, accessKey=None, secretAccessKey=None)
spark.executeAthena(connName, region=None, query=None, options=None, workGroup=None,
    catalog=None, sessionToken=None, accessKey=None, secretAccessKey=None)
spark.writeAthena(df, connName, tableName, mode, options=None, region=None,
    workGroup=None, catalog=None, sessionToken=None, accessKey=None,
    secretAccessKey=None, path=None, createTable=None, replaceTable=None,
    repairTable=None)
# DATABRICKS
spark.executeDatabricks(connName, query, options=None)
spark.writeDatabricks(df, connName, tableName, mode='append', options=None)
spark.executeDatabricksStatement(connName, query)
# TRINO
spark.executeTrino(connName, query, options=None)
spark.writeTrino(df, connName, tableName, mode='append', options=None)
spark.executeTrinoStatement(connName, query)
# PRESTO
spark.executePresto(connName, query, options=None)
spark.writePresto(df, connName, tableName, mode='append', options=None)
spark.executePrestoStatement(connName, query)
# VERTICA
spark.executeVertica(connName, query, options=None)
spark.writeVertica(df, connName, tableName, mode='append', options=None)
spark.executeVerticaStatement(connName, query)
# EXASOL
spark.executeExasol(connName, query, fingerprint=None)
spark.writeExasol(df, connName, tableName, fingerprint=None, mode='append',
    options=None)
# DORIS
spark.readDoris(connName, tableIdentifier=None, options=None)
spark.executeDoris(connName, tableIdentifier=None, filterQuery=None, options=None)
spark.writeDoris(df, connName, tableIdentifier=None, mode='append', options=None)
# DREMIO
spark.executeDremio(connName, query, options=None)
spark.writeDremio(df, connName, tableName, mode='append', options=None)
spark.executeDremioStatement(connName, query)
# DREMIOFLIGHT
spark.executeDremioFlight(connName, query, options=None)
spark.writeDremioFlight(df, connName, tableName, mode='append', options=None)
spark.executeDremioFlightStatement(connName, query)
# FIREBOLT
spark.executeFirebolt(connName, sqlQuery)
spark.readFirebolt(connName, sqlQuery)
spark.writeFirebolt(df, connName, tableName, mode='append', options=None)
spark.executeFireboltSQL(connName, sqlStatement)
spark.listFireboltTables(connName)
# SINGLESTORE
spark.readSingleStore(connName, query, options=None)
spark.readSingleStoreTable(connName, tableName, options=None)
spark.writeSingleStore(df, connName, tableName, mode, options=None)
spark.executeSingleStore(connName, query, options=None)
# MOTHERDUCK
spark.executeMotherDuck(connName, query)
spark.readMotherDuck(connName, query)
spark.executeMotherDuckSQL(connName, statement)
spark.writeMotherDuck(df, connName, tableName, mode='append')

NoSQL, search and caches#

# MONGODB
spark.readMongodb(connName, databaseName, collectionName, options=None)
spark.writeMongodb(df, connName, databaseName, collectionName, mode)
# DYNAMODB
spark.readDynamoDB(connName, tableName, options=None, schema=None)
spark.readDynamoDBIndex(connName, tableName, indexName, options=None, schema=None)
spark.queryDynamoDB(connName, tableName, hashKeyName, hashKeyValue, rangeKeyName=None,
    rangeKeyValue=None, rangeKeyOperator='=', options=None, schema=None)
spark.writeDynamoDB(df, connName, tableName, mode='append', options=None)
spark.updateDynamoDB(df, connName, tableName, mode='append', options=None)
spark.deleteDynamoDB(df, connName, tableName, mode='append', options=None)
# COSMOSDB
spark.readCosmosDb(connName)
spark.writeCosmosDb(df, connName, mode)
# COUCHBASE
spark.readCouchbase(connName, scope=None, collection=None, service='query', **kwargs)
spark.writeCouchbase(df, connName, scope=None, collection=None, service='query',
    mode='overwrite', **kwargs)
# CASSANDRA
spark.executeCassandra(connName, tableName, query, isQuery=False)
spark.writeCassandra(df, connName, table, mode)
spark.executeCassandraStatement(connName, statement)
# SCYLLADBCLOUD
spark.readScyllaDbCloud(connName, tableName, keyspace=None, options=None)
spark.executeScyllaDbCloud(connName, query, keyspace=None)
spark.writeScyllaDbCloud(df, connName, tableName, keyspace=None, mode='append',
    options=None)
spark.executeScyllaDbCloudCQL(connName, cqlStatement, keyspace=None)
spark.listScyllaDbCloudTables(connName, keyspace=None)
spark.createScyllaDbCloudTable(connName, tableName, columns, keyspace=None)
# ELASTICSEARCH
spark.readElasticsearch(connName, indexName)
spark.writeElasticsearch(df, connName, indexName, mode)
# REDIS
spark.readRedis(connName, table=None, options=None)
spark.writeRedis(df, connName, table=None, mode='append', options=None)
# NEO4J
spark.executeNeo4j(connName, query, options=None)
spark.writeNeo4j(df, connName, nodeLabel=None, relationship=None, mode='overwrite',
    options=None)
# INFLUXDB
spark.executeInfluxDB(connName, query, database=None)
spark.writeInfluxDB(df, connName, measurement, timeColumn='time', tagColumns=None,
    fieldColumns=None, batchSize=5000, timeUnit='ns', useForeachPartition=True,
    maxPartitions=None)
spark.writeInfluxDBLineProtocol(df, connName, batchSize=5000, maxPartitions=None)

Streams#

# KAFKA
spark.readKafka(connName, topic, startingOffsets=None, options=None)
spark.writeKafka(df, connName, topic, checkpointLocation, outputMode, triggerTime,
    options=None, s3AccessKey=None, s3SecretKey=None, s3SessionToken=None,
    s3Region=None, s3Endpoint=None)
# KINESIS
spark.readKinesisStream(connName, startingPosition, consumerType, options=None)
spark.writeKinesisStream(connName, df, checkpointLocation, outputMode, triggerTime,
    outputType, options=None, forEachBatch=None)
# EVENTHUBS
spark.readAzureEventHubs(connName, consumerGroup, **kwargs)
spark.writeAzureEventHubs(df, connName, **kwargs)
# PULSAR
spark.readPulsar(connName, topic=None, options=None)
spark.readPulsarStream(connName, topic=None, options=None)
spark.writePulsar(df, connName, topic=None, mode='append', options=None)
spark.writePulsarStream(df, connName, checkpointLocation, topic=None,
    outputMode='append', triggerTime='1 seconds', options=None)

SaaS and APIs#

# REST
spark.readREST(connName, endpoint=None, method=None, params=None, data=None,
    options=None)
spark.writeREST(df, connName, endpoint=None, method='POST', chunkSize=1000,
    options=None)
spark.readRESTArrow(connName, endpoint=None, method=None, params=None, data=None,
    options=None, columns=None)
spark.writeRESTArrow(df, connName, endpoint=None, method='POST', chunkSize=1000,
    options=None)
# SALESFORCE
spark.executeSalesforce(connName, soqlQuery, includeDeleted=False)
spark.readSalesforce(connName, soqlQuery, includeDeleted=False)
spark.writeSalesforce(df, connName, objectName, operation='insert', batchSize=2000,
    useBulk=False, externalIdField='Id', dropNulls=False, useSerial=False,
    raiseOnError=True)
spark.executeSalesforceArrow(connName, soqlQuery, includeDeleted=False, columns=None,
    options=None)
spark.readSalesforceArrow(connName, soqlQuery, includeDeleted=False, columns=None,
    options=None)
spark.writeSalesforceArrow(df, connName, objectName, operation='insert', batchSize=200,
    externalIdField='Id', options=None, **legacyOpts)
# HUBSPOT
spark.readHubSpot(connName, objectType, properties=None, limit=None, offset=None,
    filters=None, searchTerm=None, rateLimit=10)
spark.writeHubSpot(df, connName, objectType, operation='create', batchSize=100,
    rateLimit=10)
spark.readHubSpotArrow(connName, objectType, properties=None, limit=None, filters=None,
    searchTerm=None, rateLimit=None, columns=None, options=None)
spark.writeHubSpotArrow(df, connName, objectType, operation='create', batchSize=100,
    rateLimit=None, idProperty=None, options=None)
# SERVICENOW
spark.readServiceNow(connName, table, query=None, limit=1000, displayValue=True,
    options=None)
spark.executeServiceNow(connName, sourceQuery, limit=1000, options=None)
spark.writeServiceNow(df, connName, table, method='POST', batchSize=100)
spark.readServiceNowArrow(connName, table, query=None, limit=None, displayValue=None,
    columns=None, options=None)
spark.executeServiceNowArrow(connName, sourceQuery, columns=None, options=None)
spark.writeServiceNowArrow(df, connName, table, operation='create', idColumn='sys_id',
    options=None)
# SHOPIFY
spark.readShopify(connName, resource, params=None, maxResults=None,
    shopifyTier='enterprise', rateLimit=None, batchSize=250)
spark.writeShopify(df, connName, resource, operation='create', batchSize=50,
    shopifyTier='enterprise', rateLimit=None)
spark.readShopifyArrow(connName, resource, params=None, maxResults=None, rateLimit=None,
    batchSize=250, recordsPath=None, columns=None, options=None)
spark.writeShopifyArrow(df, connName, resource, operation='create', batchSize=50,
    wrapperKey=None, options=None)
# ZOHOCRM
spark.readZohoCRM(connName, module, perPage=200, maxPages=None)
spark.executeZohoCRM(connName, module)
spark.readZohoCRMArrow(connName, module, perPage=None, maxPages=None, columns=None,
    options=None)
spark.executeZohoCRMArrow(connName, module, columns=None, options=None)
# M365BC
spark.readM365BC(connName, entity, operation='get_all', params=None)
spark.writeM365BC(df, connName, entity, operation='create', batchSize=100)
spark.readM365BCArrow(connName, entity, params=None, columns=None, options=None)
spark.writeM365BCArrow(df, connName, entity, operation='create', batchSize=100,
    options=None)
# SPLUNK
spark.executeSplunk(connName, searchQuery, operation='get_all', params=None,
    requestTimeout=30)
spark.writeSplunk(df, connName, batchSize=100, index=None, source=None, sourcetype=None,
    hecUrl=None, hecHost=None, hecPort=None, hecPath='/services/collector/event',
    requestTimeout=30, requestChannel=None)
spark.executeSplunkArrow(connName, searchQuery, earliestTime=None, latestTime=None,
    requestTimeout=None, columns=None, options=None)
spark.writeSplunkArrow(df, connName, index=None, source=None, sourcetype=None,
    hecUrl=None, hecHost=None, hecPort=8088, hecPath='/services/collector/event',
    requestChannel=None, options=None)

Vector stores#

# PINECONE
spark.readPinecone(connName, ids_to_fetch=None, schema=None, options=None)
spark.writePinecone(df, connName=None, options=None, mode='upsert')
# QDRANT
spark.readQdrant(connName, collectionName=None, schema=None, options=None)
spark.writeQdrant(df, connName, collectionName=None, options=None)
# WEAVIATECLOUD
spark.createWeaviateCloudClass(connName, className, properties, vectorizer='none',
    moduleConfig=None)
spark.readWeaviateCloud(connName, className, properties=None, where_filter=None,
    limit=None, batchSize=10000)
spark.writeWeaviateCloud(df, connName, className, batchSize=100, maxPartitions=None,
    idColumn=None, vectorColumn=None, mode='append', options=None, writer='spark')
spark.listWeaviateCloudClasses(connName)
spark.deleteWeaviateCloudObject(connName, className, objectId)

Model Context Protocol#

# MCP
spark.listMcpTools(connName)
spark.callMcpTool(connName, toolName, arguments=None)
spark.listMcpResources(connName)
spark.readMcpResource(connName, uri)
spark.listMcpPrompts(connName)
spark.getMcpPrompt(connName, promptName, arguments=None)
spark.completeWithMcpTool(llmProvider, llmConnName, mcpConnName, df, toolName,
    arguments=None, argumentsColumn=None, promptColumn='prompt', promptTemplate=None,
    mcpPromptTemplate=None, systemPrompt=None, systemPromptColumn=None,
    outputColumn='llm_response', mcpResultColumn='mcp_tool_result',
    mcpErrorColumn='mcp_tool_error', model=None, temperature=0.0, maxTokens=None,
    responseFormat=None, continueOnError=False, timeout=120, llmOptions=None)
spark.getMcpLlmConfig(connName, provider='openai', serverName=None,
    requireApproval='always', includeSecrets=False)