Ausführen von Spark-Workloads von einem Drittanbieter-Client aus¶

Hinzufügen von Verbindungen mit Snowflake CLI¶

Sie können mit Snowflake CLI Verbindungseigenschaften hinzufügen, mit denen Snowpark Connect for Spark eine Verbindung zu Snowflake herstellen kann. Ihre Änderungen werden in einer config.toml-Datei gespeichert.

1. Führen Sie den folgenden Befehl aus, um eine Verbindung über den snow connection-Befehl :add hinzuzufügen.

   snow connection add
   

   Copy

2. Folgen Sie den Aufforderungen, um eine Verbindung zu definieren.

   Geben Sie spark-connect als Verbindungsnamen an.

   Dieser Befehl fügt eine Verbindung zu Ihrer:file:config.toml-Datei hinzu, wie im folgenden Beispiel:

   [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. Führen Sie den folgenden Befehl aus, um zu prüfen, ob die Verbindung funktioniert.

   Sie können die Verbindung auf diese Weise testen, nachdem Sie sie mit Snowflake CLI hinzugefügt haben.

   snow connection list
   snow connection test --connection spark-connect
   

   Copy

Hinzufügen einer Verbindung durch manuelles Schreiben einer Verbindungsdatei¶

Sie können eine connections.toml-Datei manuell schreiben oder aktualisieren, damit Ihr Code eine Verbindung zu Snowpark Connect for Spark auf Snowflake herstellen kann.

1. Führen Sie den folgenden Befehl aus, damit Ihre connections.toml-Datei nur dem Eigentümer (Benutzer) Lese- und Schreibzugriff gewährt.

   chmod 0600 "~/.snowflake/connections.toml"
   

   Copy

2. Bearbeiten Sie die connections.toml-Datei so, dass sie eine [spark-connect]-Verbindung mit den Verbindungseigenschaften im folgenden Beispiel enthält.

   Ersetzen Sie die Werte durch Ihre eigenen Verbindungsspezifikationen.

   [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¶

Anweisungen zum Einrichten Ihrer Python-Umgebung und zum Installieren von Snowpark Connect finden Sie in der offiziellen Snowpark Connect-Dokumentation:

Zu den wichtigsten Schritten gehören:

1. Creating a Python virtual environment

2. Installing Snowpark Connect for Spark

3. Einrichten Ihrer Verbindungskonfiguration

Schritt 2: Erstellen eines Snowpark Connect Server-Skripts¶

Erstellen Sie ein Python-Skript, um den Snowpark Connect-Server zu starten.

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. Starten des Snowpark Connect-Servers

   # 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

Schritt 3: Einrichten der Scala-Anwendung¶

1. Fügen Sie die Spark Connect-Clientabhängigkeit Ihrer „build.sbt“-Datei hinzu:

   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. Führen Sie Scala-Code aus, um eine Verbindung mit dem Snowpark Connect for Spark-Server herzustellen.

   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. Kompilieren Sie Ihre Anwendung, und führen Sie sie aus.

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

   Copy