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

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

Installer Snowpark Connect for Spark¶

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

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

   Copy

2. Installez le paquet Snowpark Connect for Spark.

   pip install --upgrade --force-reinstall 'snowpark-connect[jdk]'
   

   Copy

3. 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 snowflake.snowpark_connect
   
   # Import snowpark_connect *before* importing pyspark libraries
   from pyspark.sql.types import Row
   
   spark = snowflake.snowpark_connect.server.init_spark_session()
   

   Copy

Step 1: Set up your Snowpark Connect for Spark environment¶

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

1. Créer un environnement virtuel Python et installer Snowpark Connect.

2. Set up a connection.

Step 2: Create a Snowpark Connect for Spark server script and launch the server¶

1. Create a Python script to launch the Snowpark Connect for Spark server.

   # 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. Launch the Snowpark Connect for Spark server.

   # 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

Step 3: Set up your Scala application¶

1. Add the Spark Connect client dependency to your build.sbt file.

   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. Execute Scala code to connect to the Snowpark Connect for Spark server.

   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

Scala UDF support on Snowpark Connect for Spark¶

When using user-defined functions or custom code, do one of the following:

• Register a class finder to monitor and upload class files.

  import org.apache.spark.sql.connect.client.REPLClassDirMonitor
  
  val classFinder = new REPLClassDirMonitor("/absolute/path/to/target/scala-2.12/classes")
  spark.registerClassFinder(classFinder)
  

  Copy

• Chargez les dépendances JAR si nécessaire.

  spark.addArtifact("/absolute/path/to/dependency.jar")
  

  Copy

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

With the following list of checks, you can troubleshoot Snowpark Connect for Spark installation and use.

• Assurez-vous que Java et Python sont basés sur la même architecture.

• Use the most recent Snowpark Connect for Spark package file, as described in 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
  

  Copy

É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"
  

  Copy

Step 2: Setup and Authentication¶

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

   Copy

2. 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';
   

   Copy

Step 3: Connect to Spark Connect server¶

• 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())
  

  Copy