Exécuter des charges de travail Spark à partir d’un client tiers¶

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.

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

   snow connection add
   

   Copy

2. 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"
   

   Copy

3. 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
   

   Copy

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.

1. 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"
   

   Copy

2. 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"
   

   Copy

Step 1: Set Up Snowark Connect Environment¶

Pour obtenir des instructions sur la configuration de votre environnement Python et l’installation de Snowpark Connect, suivez la documentation officielle de Snowpark Connect :

Les étapes clés sont les suivantes :

1. Creating a Python virtual environment

2. Installing Snowpark Connect for Spark

3. Configuration de votre configuration de connexion

Étape 2 : Créer un script pour le serveur Snowpark Connect¶

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

1. 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()
   

   Copy

2. Lancer le serveur Snowpark Connect

   # Make sure you're in the correct Python environment
   pyenv activate your-snowpark-connect-env
   
   # Run the server script
   python launch-snowpark-connect.py
   

   Copy

Étape 3 : Configurer l’application Scala¶

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

   libraryDependencies += "org.apache.spark" %% "spark-connect-client-jvm" % "3.5.3"
   
   // Add JVM options for Java 9+ module system compatibility
   javaOptions ++= Seq(
     "--add-opens=java.base/java.nio=ALL-UNNAMED"
   )
   

   Copy

2. Exécuter du code Scala pour se 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()
       }
     }
   }
   

   Copy

3. Compilez et exécutez votre application.

   # Compile your Scala application
   sbt compile
   
   # Run the application
   sbt "runMain SnowparkConnectExample"
   

   Copy