Sesiones de Spark

Una sesión de Spark le da toda la API de DataFrames de PySpark sobre sus tablas de Nexus, mediante Spark Connect. Úsela cuando una sentencia SQL no basta: explorar en un notebook, encadenar transformaciones o preparar un DataFrame para ingest_frame.

Para una sola consulta, nexus.sql() es más simple y más barato. Para scripts sin supervisión, use un trabajo.

Antes de empezarCopiar el enlace

• Usted es administrador del espacio de trabajo. Una sesión lee el almacenamiento directamente, más allá del acceso a carpetas, así que cualquier otra persona recibe RefusedError.
• Instale el extra de Spark. Fija la versión de PySpark que usa Nexus; otras versiones fallan de formas confusas.

pip install "klarety[spark]"

• El espacio de trabajo tiene créditos para la sesión.

Elija un rolCopiar el enlace

Una sesión se abre con un rol:

Pida lo mínimo que necesite. Pedir un rol que su clave no tiene falla con AuthError (401).

Abra, use y cierre una sesiónCopiar el enlace

Cerrar una sesión requiere su ID. nexus.spark() devuelve un SparkSession listo pero no el ID, así que esa sesión corre hasta el límite de su plan. Para cerrarla al terminar (el caso habitual), ábrala mediante la API, que devuelve el ID:

import os

import httpx
from pyspark.sql import SparkSession

API = "https://api.klarety.ai"

token = httpx.post(
    f"{API}/v1/token", json={"api_key": os.environ["KLARETY_API_KEY"]}, timeout=30
).raise_for_status().json()["access_token"]
headers = {"Authorization": f"Bearer {token}"}

opened = httpx.post(
    f"{API}/v1/spark/session", json={"role": "read"}, headers=headers, timeout=360
).raise_for_status().json()

spark = SparkSession.builder.remote(opened["connect_url"]).getOrCreate()
try:
    orders = spark.table("orders")
    orders.groupBy("customer").count().orderBy("count", ascending=False).show(10)
finally:
    spark.stop()
    httpx.delete(
        f"{API}/v1/spark/session/{opened['session_id']}", headers=headers, timeout=30
    ).raise_for_status()
    print("Session closed")

El bloque finally cierra la sesión aunque su código falle. Abrirla puede tardar unos minutos, de ahí el tiempo de espera largo. La respuesta también trae expires_at, cuando la conexión deja de aceptarse.

El atajo del SDKCopiar el enlace

Para una mirada rápida en la que la sesión puede correr hasta el límite de su plan, el SDK la abre en una línea:

from klarety import KlaretyError, Nexus

with Nexus.connect() as nexus:
    try:
        spark = nexus.spark(role="read")
    except KlaretyError as refused:
        raise SystemExit(str(refused))
    spark.table("orders").show(5)

nexus.spark() se niega antes de abrir nada si falta PySpark ("install klarety[spark] to open a Spark session") o si su versión es distinta de la de Nexus, nombrando ambas versiones; reinstale con pip install "klarety[spark]".

Vea qué hace SparkCopiar el enlace

with Nexus.connect() as nexus:
    print(nexus.spark_status())  # off, starting, ready or busy
    print(nexus.spark_usage())   # your machine's CPUs and memory, and their use over the last hour

Cargue un DataFrame en una tablaCopiar el enlace

nexus.ingest_frame() escribe un DataFrame en una tabla de Nexus con la misma limpieza que una carga de archivo:

run = nexus.ingest_frame(summary_frame, table="customer_summary", key=["customer"])
print(run.run_id, run.state)

Con key, las filas existentes se actualizan en vez de agregarse dos veces. Siga la carga con nexus.ingest_status(run.run_id), como en Cargar datos.

Qué puede salir malCopiar el enlace

Próximos pasosCopiar el enlace

• Ejecutar SQL para consultas sueltas.
• Trabajos y trabajos por lotes para correr un script en segundo plano.
• Límites y límites de solicitudes para leer los límites de sesión de su plan.