サードパーティクライアントからSparkワークロードを実行¶

Snowflake CLI を使用して接続を追加します。¶

Snowpark Connect for Spark がSnowflakeに接続するために使える接続プロパティを追加するために、 Snowflake CLI を使用できます。変更は config.toml ファイルに保存されます。

1. snow connection コマンドを使用して接続を追加するには、次のコマンドを実行します：追加 コマンド

   snow connection add
   

   Copy

2. プロンプトに従って接続を定義します。

   必ず接続名に spark-connect を指定してください。

   このコマンドは次の例のように、 config.toml ファイルに接続を追加します。

   [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. 以下のコマンドを実行して、接続が機能することを確認します。

   Snowflake CLI を使用して接続を追加した場合は、この方法でテストできます。

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

   Copy

接続ファイルを手動で書き込んで接続を追加する¶

Snowflake上でコードが Snowpark Connect for Spark に接続できるように、 connections.toml ファイルを手動で記述または更新できます。

1. 次のコマンドを実行して、 connections.toml ファイルでは、所有者（ユーザー）のみが読み取りおよび書き込みアクセスできることを確認します。

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

   Copy

2. 次の例の接続プロパティを使用した [spark-connect] 接続が含まれるように、connections.toml ファイルを編集します。

   値を独自の接続仕様に置き換えてください。

   [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

Snowpark Connect for Spark をインストールする¶

Pythonパッケージとして Snowpark Connect for Spark をインストールできます。

1. Pythonの仮想環境を作成します。

   たとえば、次の例のようにCondaを使用できます。

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

   Copy

2. Snowpark Connect for Spark パッケージをインストールします。

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

   Copy

3. Snowpark Connect for Spark サーバーを開始して Snowpark Connect for Spark セッションを作成するには、Pythonコードを追加します。

   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