서드 파티 클라이언트에서 Spark 워크로드 실행하기

Jupyter Notebooks, VS Code, Spark 클러스터를 관리할 필요가 없는 Python 기반 인터페이스에서 대화형으로 Spark 워크로드를 실행할 수 있습니다. 워크로드는 Snowflake 인프라에서 실행됩니다.

예를 들면, 다음 작업을 수행할 수 있습니다.

  1. 전제 조건을 갖추고 있는지 확인합니다.

  2. Snowflake에서 |spconnect|와 연결할 수 있는 환경을 설정합니다.

  3. Snowpark Connect for Spark 를 설치합니다.

  4. 클라이언트에서 PySpark 코드를 실행하여 Snowflake에서 실행합니다.

전제 조건

Python 및 Java 설치가 동일한 컴퓨터 아키텍처에 기반을 두고 있는지 확인합니다. 예를 들어, Python이 arm64 기반인 경우 Java도 arm64여야 합니다(예: x86_64 아님).

환경 설정하기

코드로 Snowflake에서 |spconnect|에 연결할 수 있도록 하여 개발 환경을 설정할 수 있습니다. Snowflake 클라이언트 코드에 연결하려면, 연결 세부 정보가 포함된 .toml 파일을 사용합니다.

|sf-cli|가 설치되어 있다면 이를 사용하여 연결을 정의할 수 있습니다. 그렇지 않으면 config.toml 파일에 연결 매개 변수를 수동으로 작성할 수 있습니다.

|sf-cli|를 사용하여 연결 추가하기

|sf-cli|를 사용하면 |spconnect|에서 Snowflake에 연결할 수 있는 연결 속성을 추가할 수 있습니다. 변경 사항은 config.toml 파일에 저장됩니다.

  1. 다음 명령을 실행해 snow connection add 명령을 사용하는 연결을 추가합니다.

    snow connection add
    
    Copy
  2. 프롬프트에 따라 연결을 정의합니다.

    :code:`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. 다음 명령을 실행하여 연결이 작동하는지 확인합니다.

    |sf-cli|를 사용하여 추가한 경우 이 방법으로 연결을 테스트할 수 있습니다.

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

연결 파일을 수동으로 작성하여 연결 추가하기

코드로 Snowflake에서 |spconnect|에 연결할 수 있도록 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 패키지로 |spconnect|를 설치할 수 있습니다.

  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. Python 코드를 추가하여 Snowpark Connect for Spark 서버를 시작하고 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

Run Python code from your client

Once you have an authenticated connection in place, you can write code as you normally would.

PySpark 클라이언트 라이브러리를 사용하여 |spconnect|에 연결되는 PySpark 코드를 실행할 수 있습니다.

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

Scala 애플리케이션용 Snowpark Connect 시작하기

이 가이드에서는 Snowpark Connect를 설정하고 Scala 애플리케이션을 Snowpark Connect 서버에 연결하는 방법을 안내합니다.

Step 1: Set Up Snowark Connect Environment

Python 환경 설정 및 Snowpark Connect 설치에 대한 지침은 공식 Snowpark Connect 설명서를 참조하세요.

주요 단계는 다음과 같습니다.

  1. Creating a Python virtual environment

  2. Installing Snowpark Connect for Spark

  3. 연결 구성 설정

2단계: Snowpark Connect 서버 스크립트 생성

Snowpark Connect 서버를 시작하는 Python 스크립트를 만듭니다.

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

3단계: Scala 애플리케이션 설정

  1. build.sbt 파일에 Spark Connect 클라이언트 종속성을 추가합니다.

    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. Scala Code를 실행하여 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. 애플리케이션을 컴파일하고 실행합니다.

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

|spconnect|에 대한 UDF 지원

사용자 정의 함수 또는 사용자 지정 코드를 사용할 때 다음 중 하나를 수행합니다.

  • ClassFinder를 등록하여 클래스 파일을 모니터링하고 업로드합니다.

    import org.apache.spark.sql.connect.client.REPLClassDirMonitor
    
    val classFinder = new REPLClassDirMonitor("/absolute/path/to/target/scala-2.12/classes")
    spark.registerClassFinder(classFinder)
    
    Copy
  • 필요한 경우 JAR 종속성을 업로드합니다.

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

Snowpark Connect for Spark 설치 문제 해결하기

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

  • Java와 Python이 :ref:`동일한 아키텍처에 기반<label-snowpark_connect_jupyter_prereq>`을 두고 있는지 확인합니다.

  • Use the most recent Snowpark Connect for Spark package file, as described in Snowpark Connect for Spark 설치하기.

  • PySpark 코드가 포함된 python 명령이 로컬 실행, 즉 Snowflake 연결 없이 올바르게 작동하는지 확인합니다.

    예를 들면, 다음과 같은 명령을 실행합니다.

    python your_pyspark_file.py
    
    Copy