Spark Connector 시작하기

종속성 관리

Spark 코어, Spark SQL 및 MongoDB Spark Connector 종속성을 종속성 관리 도구에 제공하십시오.

버전 3.2.0 부터 시작됩니다. Apache Spark 는 Scala 2.12 과 2.13을 모두 지원합니다. Spark 3.1.3 및 이전 버전은 Scala 2 만 지원합니다.12. Scala 의 10.7. 버전은 두 버전 모두에 대한 지원을 제공하기0 Spark Connector 위해 두 개의 아티팩트를 생성합니다.

• org.mongodb.spark:mongo-spark-connector_2.12:10.7.0 Scala 2.12에 대해 컴파일되고 Spark 3.1.x 이상을 지원합니다.

• org.mongodb.spark:mongo-spark-connector_2.13:10.7.0 Scala 2.13에 대해 컴파일되고 Spark 3.2.x 이상을 지원합니다.

다음 Maven pom.xml 파일 발췌문은 Scala 2.12과 호환되는 종속성을 포함하는 방법을 보여줍니다.

<dependencies>
  <dependency>
    <groupId>org.mongodb.spark</groupId>
    <artifactId>mongo-spark-connector_2.12</artifactId>
    <version>10.7.0</version>
  </dependency>
  <dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-core_2.12</artifactId>
    <version>3.3.1</version>
  </dependency>
  <dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-sql_2.12</artifactId>
    <version>3.3.1</version>
  </dependency>
</dependencies>

구성

SparkSession를 통해 connector 구성을 지정할 때는 설정 접두사를 적절히 붙여야 합니다. 자세한 내용 및 기타 사용 가능한 MongoDB Spark Connector 옵션은 Spark 구성 가이드를 참조하세요.

package com.mongodb.spark_examples;
import org.apache.spark.sql.SparkSession;
public final class GettingStarted {
  public static void main(final String[] args) throws InterruptedException {
    /* Create the SparkSession.
     * If config arguments are passed from the command line using --conf,
     * parse args for the values to set.
     */
    SparkSession spark = SparkSession.builder()
      .master("local")
      .appName("MongoSparkConnectorIntro")
      .config("spark.mongodb.read.connection.uri", "mongodb://127.0.0.1/test.myCollection")
      .config("spark.mongodb.write.connection.uri", "mongodb://127.0.0.1/test.myCollection")
      .getOrCreate();
    // Application logic
  }
}

• spark.mongodb.read.connection.uri 은 MongoDB 서버 주소(127.0.0.1), 연결할 데이터베이스 (test), 데이터를 읽을 컬렉션 (myCollection) 및 읽기 설정 (read preference) 을 지정합니다.

• spark.mongodb.write.connection.uri 은 MongoDB 서버 주소(127.0.0.1), 연결할 데이터베이스(test) 및 데이터를 쓸 컬렉션(myCollection)을 지정합니다.

SparkSession 객체를 사용하여 MongoDB에 데이터를 쓰고, MongoDB에서 데이터를 읽고, 데이터 세트를 만들고, SQL 작업을 수행할 수 있습니다.

Python Spark 셸

이 튜토리얼에서는 pyspark 셸을 사용하지만 코드는 독립형 Python 애플리케이션에서도 작동합니다.

pyspark 셸을 시작할 때 다음을 지정할 수 있습니다.

• --packages 옵션을 클릭하여 MongoDB Spark Connector 패키지를 다운로드합니다. 다음 패키지를 이용할 수 있습니다.

  • mongo-spark-connector

• --conf 옵션을 사용하여 MongoDB Spark Connector를 구성합니다. 이 설정은 SparkConf 객체를 구성합니다.

  참고

  SparkConf 를 사용하여 Spark Connector 를 구성하는 경우 설정 접두사를 적절히 붙여야 합니다. 자세한 내용 및 기타 사용 가능한 MongoDB Spark Connector 옵션은 Spark 구성 가이드 를 참조하세요.

다음은 명령줄에서 pyspark 셸을 시작하는 예시입니다.

./bin/pyspark --conf "spark.mongodb.read.connection.uri=mongodb://127.0.0.1/test.myCollection?readPreference=primaryPreferred" \
              --conf "spark.mongodb.write.connection.uri=mongodb://127.0.0.1/test.myCollection" \
              --packages org.mongodb.spark:mongo-spark-connector_2.12:10.7.0

• spark.mongodb.read.connection.uri 은 MongoDB 서버 주소(127.0.0.1), 연결할 데이터베이스(test), 데이터를 읽을 컬렉션(myCollection) 및 읽기 설정을 지정합니다.

• spark.mongodb.write.connection.uri 은 MongoDB 서버 주소 (127.0.0.1), 연결할 데이터베이스 (test) 및 데이터를 쓰기 (write) 컬렉션 (myCollection)을 지정합니다. 기본값 으로 포트 27017 에 연결합니다.

• packages 옵션은 Spark Connector의 Maven 좌표를 groupId:artifactId:version 형식으로 지정합니다.

이 튜토리얼의 예시에서는 이 데이터베이스와 컬렉션을 사용합니다.

SparkSession 객체 만들기

spark.mongodb.read.connection.uri 및 spark.mongodb.write.connection.uri 구성 옵션을 지정하여 pyspark를 시작하는 경우, 기본 SparkSession 객체가 해당 옵션을 사용합니다. pyspark 내에서 고유한 SparkSession 객체를 만들려는 경우 SparkSession.builder를 사용하고 다른 구성 옵션을 지정할 수 있습니다.

from pyspark.sql import SparkSession
my_spark = SparkSession \
    .builder \
    .appName("myApp") \
    .config("spark.mongodb.read.connection.uri", "mongodb://127.0.0.1/test.coll") \
    .config("spark.mongodb.write.connection.uri", "mongodb://127.0.0.1/test.coll") \
    .getOrCreate()

SparkSession 객체를 사용하여 MongoDB에 데이터를 쓰고, MongoDB에서 데이터를 읽고, DataFrame을 생성하고, SQL 작업을 수행할 수 있습니다.

Spark 셸

Spark 셸을 시작할 때 다음을 지정합니다:

• --packages 옵션을 클릭하여 MongoDB Spark Connector 패키지를 다운로드합니다. 다음 패키지를 이용할 수 있습니다.

  • mongo-spark-connector

• --conf 옵션을 사용하여 MongoDB Spark Connector를 구성합니다. 이 설정은 SparkConf 객체를 구성합니다.

  참고

  SparkConf 를 사용하여 Spark Connector 를 구성하는 경우 설정 접두사를 적절히 붙여야 합니다. 자세한 내용 및 기타 사용 가능한 MongoDB Spark Connector 옵션은 Spark 구성 가이드 를 참조하세요.

예를 들면 다음과 같습니다.

./bin/spark-shell --conf "spark.mongodb.read.connection.uri=mongodb://127.0.0.1/test.myCollection?readPreference=primaryPreferred" \
                  --conf "spark.mongodb.write.connection.uri=mongodb://127.0.0.1/test.myCollection" \
                  --packages org.mongodb.spark:mongo-spark-connector_2.12:10.7.0

• spark.mongodb.read.connection.uri 은 MongoDB 서버 주소(127.0.0.1), 연결할 데이터베이스(test), 데이터를 읽을 컬렉션(myCollection) 및 읽기 설정을 지정합니다.

• spark.mongodb.write.connection.uri 은 MongoDB 서버 주소 (127.0.0.1), 연결할 데이터베이스 (test) 및 데이터를 쓰기 (write) 컬렉션 (myCollection)을 지정합니다. 기본값 으로 포트 27017 에 연결합니다.

• packages 옵션은 Spark Connector의 Maven 좌표를 groupId:artifactId:version 형식으로 지정합니다.

import com.mongodb.spark._

독립형 Scala 애플리케이션

scalaVersion := "2.12",
libraryDependencies ++= Seq(
  "org.mongodb.spark" %% "mongo-spark-connector_2.12" % "10.7.0",
  "org.apache.spark" %% "spark-core" % "3.3.1",
  "org.apache.spark" %% "spark-sql" % "3.3.1"
)

package com.mongodb
object GettingStarted {
  def main(args: Array[String]): Unit = {
    /* Create the SparkSession.
     * If config arguments are passed from the command line using --conf,
     * parse args for the values to set.
     */
    import org.apache.spark.sql.SparkSession
    val spark = SparkSession.builder()
      .master("local")
      .appName("MongoSparkConnectorIntro")
      .config("spark.mongodb.read.connection.uri", "mongodb://127.0.0.1/test.myCollection")
      .config("spark.mongodb.write.connection.uri", "mongodb://127.0.0.1/test.myCollection")
      .getOrCreate()
  }
}

문제 해결

java.net.BindException: Can't assign requested address을 받을 경우

• 다른 Spark 셸이 이미 실행 중이 아닌지 확인합니다.

• SPARK_LOCAL_IP 환경 변수를 설정합니다. 예:

  export SPARK_LOCAL_IP=127.0.0.1

• Spark 셸을 시작할 때 다음 옵션을 포함합니다:

  --driver-java-options "-Djava.net.preferIPv4Stack=true"

이 튜토리얼의 예시를 실행하는 중에 오류가 발생하는 경우 로컬 Ivy 캐시(~/.ivy2/cache/org.mongodb.spark 및 ~/.ivy2/jars)를 지워야 할 수 있습니다.