Exécuter des charges de travail Spark à partir de VS Code, Jupyter Notebooks ou un terminal¶

Vous pouvez exécuter des charges de travail Spark de manière interactive à partir de Jupyter Notebooks, VS Code, ou toute interface basée sur Python sans avoir besoin de gérer un cluster Spark. Les charges de travail s’exécutent sur l’infrastructure Snowflake.

Par exemple, vous pouvez effectuer les tâches suivantes :

Confirmer que vous avez des conditions préalables.
Configurer votre environnement pour vous connecter avec Snowpark Connect for Spark sur Snowflake.
Installez Snowpark Connect for Spark.
Exécuter le code PySpark depuis votre client pour l’exécuter sur Snowflake.

Conditions préalables¶

Vérifiez que vos installations Python et Java sont basées sur la même architecture informatique. Par exemple, si Python est basé sur arm64, Java doit également être basé sur arm64 (et non x86_64, par exemple).

Configurer votre environnement¶

Vous pouvez configurer votre environnement de développement en veillant à ce que le code puisse se connecter à Snowpark Connect for Spark sur Snowflake. Pour se connecter à Snowflake, le code client utilisera un fichier .toml contenant les détails de la connexion.

Si Snowflake CLI est installée, vous pouvez l’utiliser pour définir une connexion. Sinon, vous pouvez écrire manuellement les paramètres de connexion dans un fichier config.toml.

Ajouter une connexion en utilisant Snowflake CLI¶

Vous pouvez utiliser Snowflake CLI pour ajouter des propriétés de connexion que Snowpark Connect for Spark peut utiliser pour se connecter à Snowflake. Vos modifications sont enregistrées dans un fichier config.toml.

Exécutez la commande suivante pour ajouter une connexion à l’aide de la commande :command: add snow connection.
```
snow connection add
```

Suivez les invites pour définir une connexion.

Veillez à spécifier spark-connect comme nom de connexion.

Cette commande ajoute une connexion à votre fichier config.toml, comme dans l’exemple suivant :

[connections.spark-connect]
host = "example.snowflakecomputing.com"
port = 443
account = "example"
user = "test_example"
password = "password"
protocol = "https"
warehouse = "example_wh"
database = "example_db"
schema = "public"

Exécutez la commande suivante pour confirmer que la connexion fonctionne.

Vous pouvez tester la connexion de cette manière lorsque vous l’avez ajoutée en utilisant Snowflake CLI.
```
snow connection list
snow connection test --connection spark-connect
```

Ajouter une connexion en écrivant manuellement un fichier de connexion¶

Vous pouvez écrire ou mettre à jour manuellement un fichier connections.toml afin que votre code puisse se connecter à Snowpark Connect for Spark sur Snowflake.

Exécutez la commande suivante pour vous assurer que votre fichier connections.toml autorise uniquement au propriétaire (utilisateur) d’avoir un accès en lecture et en écriture.
```
chmod 0600 "~/.snowflake/connections.toml"
```
Modifiez le fichier connections.toml afin qu’il contienne une connexion [spark-connect] avec les propriétés de connexion dans l’exemple suivant.

Veillez à remplacer les valeurs par vos propres spécificités de connexion.
```
[spark-connect]
host="my_snowflake_account.snowflakecomputing.com"
account="my_snowflake_account"
user="my_user"
password="&&&&&&&&"
warehouse="my_wh"
database="my_db"
schema="public"
```

Installer Snowpark Connect for Spark¶

Vous pouvez installer Snowpark Connect for Spark en tant que paquet Python.

Créez un environnement virtuel Python.

Confirmez que votre version de Python est 3.10 ou ultérieure et antérieure à 3.13 en exécutant python3 --version.
```
python3 -m venv .venv
source .venv/bin/activate
```
Installez le paquet Snowpark Connect for Spark.
```
pip install --upgrade --force-reinstall 'snowpark-connect[jdk]'
```
Ajoutez du code Python pour démarrer un serveur Snowpark Connect for Spark et créez une session Snowpark Connect for Spark.
```
from snowflake import snowpark_connect
spark=snowpark_connect.init_spark_session()
```

Exécuter le code Python de votre client¶

Une fois que vous disposez d’une connexion authentifiée, vous pouvez écrire le code comme vous le feriez normalement.

Vous pouvez exécuter le code PySpark qui se connecte à Snowpark Connect for Spark en utilisant la bibliothèque client PySpark.

# Row is imported in the previous code snippet

df = spark.createDataFrame([
    Row(a=1, b=2.),
    Row(a=2, b=3.),
    Row(a=4, b=5.),])

print(df.count())

Exécuter du code Scala depuis votre client¶

Vous pouvez exécuter des applications Scala qui se connectent à Snowpark Connect for Spark en utilisant la bibliothèque client Spark Connect.

Ce guide vous guide à travers la configuration de Snowpark Connect et la connexion de vos applications Scala au serveur Snowpark Connect for Spark .

Étape 1 : Configurer votre environnement Snowpark Connect for Spark¶

Configurez votre environnement en utilisant les étapes décrites dans les rubriques suivantes :

Créer un environnement virtuel Python et installer Snowpark Connect.
Configurer une connexion.

Étape 2 : Créer un script de serveur Snowpark Connect for Spark et lancer le serveur¶

Créez un script Python pour lancer le serveur Snowpark Connect for Spark.

# launch-snowpark-connect.py

from snowflake import snowpark_connect

def main():
    snowpark_connect.start_session(is_daemon=False, remote_url="sc://localhost:15002")
    print("SAS started on port 15002")

if __name__ == "__main__":
    main()

Lancez le serveur Snowpark Connect for Spark.

# Make sure you're in the correct Python environment
pyenv activate your-snowpark-connect-env

# Run the server script
python launch-snowpark-connect.py

Étape 3 : Configurer votre application Scala¶

Ajoutez la dépendance du client Spark Connect à votre fichier build.sbt :

libraryDependencies += "org.apache.spark" %% "spark-connect-client-jvm" % "3.5.6"

// Add JVM options for Java 9+ module system compatibility
javaOptions ++= Seq(
  "--add-opens=java.base/java.nio=ALL-UNNAMED"
)

Exécutez du code Scala pour vous connecter au serveur Snowpark Connect for Spark.

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.connect.client.REPLClassDirMonitor

object SnowparkConnectExample {
  def main(args: Array[String]): Unit = {
    // Create Spark session with Snowpark Connect
    val spark = SparkSession.builder().remote("sc://localhost:15002").getOrCreate()

    // Register ClassFinder for UDF support (if needed)
    // val classFinder = new REPLClassDirMonitor("target/scala-2.12/classes")
    // spark.registerClassFinder(classFinder)

    try {
      // Simple DataFrame operations
      import spark.implicits._

      val data = Seq(
        (1, "Alice", 25),
        (2, "Bob", 30),
        (3, "Charlie", 35)
      )

      val df = spark.createDataFrame(data).toDF("id", "name", "age")

      println("Original DataFrame:")
      df.show()

      println("Filtered DataFrame (age > 28):")
      df.filter($"age" > 28).show()

      println("Aggregated result:")
      df.groupBy().avg("age").show()

    } finally {
      spark.stop()
    }
  }
}

Compilez et exécutez votre application.

# Compile your Scala application
sbt compile

# Run the application
sbt "runMain SnowparkConnectExample"

Prise en charge de Scala UDF sur Snowpark Connect for Spark¶

Lorsque vous utilisez des fonctions définies par l’utilisateur ou du code personnalisé, effectuez l’une des opérations suivantes :

Activer un localisateur de classe pour surveiller et charger des fichiers de classe.

import org.apache.spark.sql.connect.client.REPLClassDirMonitor

val classFinder = new REPLClassDirMonitor("/absolute/path/to/target/scala-2.12/classes")
spark.registerClassFinder(classFinder)

Chargez les dépendances JAR si nécessaire. Vous pouvez inclure le JAR de charge de travail lui-même si un localisateur de classe n’est pas utilisé.
```
spark.addArtifact("/absolute/path/to/dependency.jar")
```

Utilisez un JAR mis en zone de préparation.

spark.conf.set("snowpark.connect.udf.java.imports", "[@mystage/dependency.jar, @db.schema.stage/other_dependency.jar]")

Utilisation de Scala 2.13¶

Par défaut, Snowpark Connect for Spark utilise Scala 2.12. Les charges de travail créées avec Scala 2.13 doivent spécifier la version de Scala en utilisant l’option de configuration « snowpark.connect.scala.version ».

// Directly in the session builder
val spark = SparkSession.builder()
  .remote("sc://localhost:15002")
  .config("snowpark.connect.scala.version", "2.13")
  .getOrCreate()

// Or via session configuration
spark.conf.set("snowpark.connect.scala.version", "2.13")

Résoudre les problèmes d’installation Snowpark Connect for Spark¶

La liste de vérifications suivante vous permettra de résoudre les problèmes d’installation et d’utilisation de Snowpark Connect for Spark.

Assurez-vous que Java et Python sont basés sur la même architecture.
Utiliser le dernier fichier de package Snowpark Connect for Spark, comme décrit dans Installer Snowpark Connect for Spark.
Confirmez que la commande Python avec le code PySpark fonctionne correctement pour l’exécution locale, c’est-à-dire sans connexion Snowflake.

Par exemple, exécutez une commande telle que la suivante :
```
python your_pyspark_file.py
```

Clients open source¶

Vous pouvez utiliser des packs clients Spark de logiciels open source (OSS) standard et prêts à l’emploi, comme les clients PySpark et Spark pour Java ou Scala—à partir de vos environnements locaux préférés, dont Jupyter Notebooks et VS Code. De cette manière, vous pouvez éviter d’installer des paquets spécifiques à Snowflake.

Cela peut s’avérer utile si vous souhaitez écrire du code Spark localement et que le code utilise les ressources de calcul Snowflake et la gouvernance d’entreprise. Dans ce scénario, vous effectuez l’authentification et l’autorisation via des jetons d’accès programmatiques (PATs).

Les sections suivantes traitent de l’installation, de la configuration et de l’authentification. Vous trouverez également un exemple PySpark simple pour valider votre connexion.

Étape 1 : Installation des paquets requis¶

Installez pyspark. Vous n’avez pas besoin d’installer de paquet Snowflake.
```
pip install "pyspark[connect]>=3.5.0,<4"
```

Étape 2 : Configuration et authentification¶

Générez un jeton d’accès programmatique (PAT).

Pour plus d’informations, consultez les rubriques suivantes :
- Utilisation de jetons d’accès programmatique pour l’authentification
- ALTER USER … ADD PROGRAMMATIC ACCESS TOKEN (PAT)
L’exemple suivant ajoute un PAT nommé TEST_PAT pour l’utilisateur sysadmin et fixe l’expiration à 30 jours.
```
ALTER USER add PAT TEST_PAT ROLE_RESTRICTION = sysadmin DAYS_TO_EXPIRY = 30;
```

Trouvez votre URL hôte Snowflake Spark Connect.

Exécutez le SQL suivant dans Snowflake pour trouver le nom d’hôte de votre compte :

SELECT t.VALUE:type::VARCHAR as type,
       t.VALUE:host::VARCHAR as host,
       t.VALUE:port as port
  FROM TABLE(FLATTEN(input => PARSE_JSON(SYSTEM$ALLOWLIST()))) AS t where type = 'SNOWPARK_CONNECT';

Étape 3 : Connexion au serveur Spark Connect¶

Pour vous connecter au serveur Spark Connect, utilisez un code tel que le code suivant :

from pyspark.sql import SparkSession
import urllib.parse

# Replace with your actual PAT.
pat = urllib.parse.quote("<pat>", safe="")

# Replace with your Snowpark Connect host from the above SQL query.
snowpark_connect_host = ""

# Define database/schema/warehouse for executing your Spark session in Snowflake (recommended); otherwise, it will be resolved from your default_namespace and default_warehouse

db_name = urllib.parse.quote("TESTDB", safe="")
schema_name = urllib.parse.quote("TESTSCHEMA", safe="")
warehouse_name = urllib.parse.quote("TESTWH", safe="")

spark = SparkSession.builder.remote(f"sc://{snowpark_connect_host}/;token={pat};token_type=PAT;database={db_name};schema={schema_name};warehouse={warehouse_name}").getOrCreate()

# Spark session is ready to use. You can write regular Spark DataFrame code, as in the following example:

from pyspark.sql import Row

df = spark.createDataFrame([
    Row(a=1, b=2.),
    Row(a=2, b=3.),
    Row(a=4, b=5.),])
print(df.count())