Execução de cargas de trabalho do Spark por meio de um cliente de terceiros¶

Adicione uma conexão usando Snowflake CLI¶

Você pode usar Snowflake CLI para adicionar propriedades de conexão que Snowpark Connect for Spark pode usar para se conectar ao Snowflake. Suas alterações são salvas em um arquivo config.toml.

1. Execute o seguinte comando para adicionar uma conexão usando o comando de conexão snow: comado add.

   snow connection add
   

   Copy

2. Siga as instruções para definir uma conexão.

   Certifique-se de especificar spark-connect como o nome da conexão.

   Este comando adiciona uma conexão ao seu arquivo config.toml, como no exemplo a seguir:

   [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. Execute o seguinte comando para confirmar que a conexão funciona.

   Você poderá testar a conexão dessa maneira quando a adicionar usando Snowflake CLI.

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

   Copy

Adição de uma conexão gravando manualmente um arquivo de conexão¶

Você pode escrever ou atualizar manualmente um arquivo connections.toml para que seu código possa se conectar ao Snowpark Connect for Spark no Snowflake.

1. Execute o seguinte comando para garantir que seu arquivo connections.toml permita que somente o proprietário (usuário) tenha acesso de leitura e gravação.

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

   Copy

2. Edite o arquivo connections.toml para que ele contenha uma conexão [spark-connect] com as propriedades de conexão no exemplo a seguir.

   Certifique-se de substituir os valores por suas próprias especificações de conexão.

   [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

Instalação do Snowpark Connect for Spark¶

Você pode instalar Snowpark Connect for Spark como um pacote Python.

1. Crie um ambiente virtual Python.

   Por exemplo, você pode usar Conda, como no exemplo a seguir.

   conda create -n xxxx pip python=3.12
   conda activate xxxx
   

   Copy

2. Instale o pacote Snowpark Connect for Spark.

   pip install --upgrade --force-reinstall snowpark-connect
   

   Copy

3. Adicione o código Python para iniciar um Snowpark Connect for Spark e crie uma sessão de Snowpark Connect for Spark.

   import os
   import snowflake.snowpark
   from snowflake import snowpark_connect
   # Import snowpark_connect before importing pyspark libraries
   from pyspark.sql.types import Row
   
   os.environ["SPARK_CONNECT_MODE_ENABLED"] = "1"
   snowpark_connect.start_session()  # Start the local Snowpark Connect for Spark session
   spark = snowpark_connect.get_session()
   

   Copy

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

Set up your environment by using steps described in the following topics:

1. Create a Python virtual environment and install 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. Compile and run your 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

• Upload JAR dependencies if needed.

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

  Copy