출처

https://data-flair.training/blogs/rdd-lineage/

Spark에서 RDD 계보 정보(리니지): ToDebugString 메소드

1. 목적

기본적으로 Spark에서는 실제 데이터에도 RDD 간의 모든 종속성이 그래프에 기록됩니다. 이것이 우리가 Spark에서 계보 그래프라고 부르는 것입니다. 이 문서는 Spark 논리적 실행 계획의 RDD 계보 개념을 담고 있습니다. 또한 toDebugString 메서드로 RDD 계보 정보 그래프를 얻는 방법을 자세히 알게 될 것입니다. 먼저 Spark RDD에 대해서도 알아 보겠습니다.

2. Spark RDD 소개

Spark RDD는 "Resilient Distributed Dataset"의 약어 입니다. RDD를 Apache Spark의 기본 데이터 구조로 간주할 수 있습니다. 구체적으로 말하면 RDD는 Apache Spark의 변경 불가능한 개체 모음입니다. 이는 클러스터의 다른 노드에서 계산하는 데 도움이 됩니다.
Spark RDD의 이름을 분해 할 때 :

• 탄력성

이것은 내결함성을 의미합니다. RDD 계보 정보 그래프(DAG)를 사용하여 노드 장애로 인해 누락되거나 손상된 파티션을 다시 계산할 수 있습니다.

• 분산

데이터가 여러 노드에 있음을 의미합니다.

• 데이터 세트

작업하는 데이터의 기록일 뿐 입니다. 또한 사용자는 데이터 세트를 외부에서 불러올 수 있습니다. 예를 들어, 특정 데이터 구조가 없는 JDBC를 통한 JSON 파일, CSV 파일, 텍스트 파일 또는 데이터베이스일 수 있습니다.

당신은 Spark dataSet 튜토리얼을 읽어야 합니다.

3. RDD 계보정보(리니지) 소개

기본적으로, RDD의 평가는 자연적으로 게으릅니다. 이는 변환의 시리즈가 RDD에서 수행되지만, 바로 평가되지는 않습니다.
Spark RDD로부터 새로운 RDD를 만드는 동안, 새로운 RDD는 Spark에서 부모 RDD의 포인터를 가져옵니다. 이는 실제 데이터가 아니라 그래프에 기록된 RDD 간의 모든 종속성과 동일합니다. 우리가 계보 그래프라고 부르는 것입니다. RDD 계보는 RDD의 모든 부모 RDD의 그래프일 뿐 입니다. RDD 연산자 그래프 또는 RDD 종속성 그래프라고도 합니다. 구체적으로 말하자면 스파크에 Transformation을 적용한 결과입니다. 그런 다음 논리적 실행 계획을 생성합니다.
또한 실제 실행 계획 또는 실행 DAG를 단계의 DAG라고 합니다.
잘 이해하기 위해 Cartesian 또는 zip을 사용하여 Spark RDD 계보의 한 예부터 시작하겠습니다. 그러나 다른 연산자를 사용하여 Spark에서 RDD 그래프를 작성할 수도 있습니다.

예시

위 그림은 다음과 같은 일련의 Transformation의 결과인 RDD 그래프를 보여줍니다.

Spark의 게으른 평가

val r00 = sc.parallelize(0 to 9)
val r01 = sc.parallelize(0 to 90 by 10)
val r10 = r00 cartesian df01
val r11 = r00.map(n => (n, n))
val r12 = r00 zip df01
val r13 = r01.keyBy(_ / 20)
val r20 = Seq(r11, r12, r13).foldLeft(r10)(_ union _)

다른 예시

다음과 같은 RDD val b=a.map()이 있다고 합시다.

RDD b는 부모 RDD a에 대한 참조를 유지해야 합니다. 이것이 RDD 계보 정보(리니지)의 종류입니다.

4. RDD 계보정보(리니지)의 논리적 실행 계획

기본적으로, 논리적 실행 계획은 초기 RDD들과 함께 초기화 됩니다. 초기 RDD는 다른 RDD에 의존하지 않는 RDD 일뿐입니다. 매우 구체적으로 말하자면 이들은 참조 캐시 데이터와 독립적입니다. 또한 실행을 위해 호출된 작업의 결과를 생성하는 RDD로 끝납니다.

Spark 작업을 실행하기 위해 SparkContext가 요청될 때 실행되는 DAG라고도 말할 수 있습니다.

5. Spark에서 RDD 계보정보(리니지) 그래프를 얻기 위한 ToDebugString 메소드

Spark에서 RDD 계보정보(리니지) 그래프를 얻기 위한 몇가지 방법이 있지만, 메소드 중 하나는 toDebugString 메소드 입니다.

toDebugString: String

Spark DStream 살펴보기

기본적으로 이 방법을 사용하여 Spark RDD 계보정보(리니지) 그래프에 대해 배울 수 있습니다.

scala> val wordCount1 = sc.textFile(“README.md”).flatMap(_.split(“\\s+”)).map((_, 1)).reduceByKey(_ + _)
wordCount1: org.apache.spark.rdd.RDD[(String, Int)] = ShuffledRDD[21] at reduceByKey at <console>:24
scala> wordCount1.toDebugString
res13: String =
(2) ShuffledRDD[21] at reduceByKey at <console>:24 []
+-(2) MapPartitionsRDD[20] at map at <console>:24 []
|  MapPartitionsRDD[19] at flatMap at <console>:24 []
|  README.md MapPartitionsRDD[18] at textFile at <console>:24 []
|  README.md HadoopRDD[17] at textFile at <console>:24 []

기본적으로 여기에서 괄호() 안의 H는 각 단계에서 병렬 처리 수준을 나타내는 숫자를 나타냅니다.
예를 들어, 위 출력에서 (2) 입니다.

scala> wordCount1.getNumPartitions
res14: Int = 2

toDebugString 메서드는 action을 실행할 때 포함되며 spark.logLineage 속성이 활성화됩니다.

$ ./bin/spark-shell –conf spark.logLineage=true
scala> sc.textFile(“README.md”, 4).count
…
15/10/17 14:46:42 INFO SparkContext: Starting job: count at <console>:25
15/10/17 14:46:42 INFO SparkContext: RDD’s recursive dependencies:
(4) MapPartitionsRDD[1] at textFile at <console>:25 []
|  README.md HadoopRDD[0] at textFile at <console>:25 []

Spark 성능 조정에 대해 읽어 보세요.

그래서 이것은 Spark RDD Lineage Tutorial에 관한 것입니다. 우리의 설명이 마음에 드셨으면 좋겠습니다.

6. 결론

따라서 이 블로그를 통해 Apache Spark RDD 계보정보(리니지) 그래프의 실제 의미를 배웠습니다. 또한 Apache Spark에서 논리적 실행 계획의 풍미를 맛 보았습니다. 그러나 toDebugString 메서드도 자세히 살펴 보았습니다. 또한 Apache Spark RDD에서 모든 계보정보(리니지) 그래프 개념을 다루었습니다.

또한 궁금한 점이 있으시면 댓글란에 문의 해주세요.

Spark를 배우려면 인기 도서를 참조하십시오.

출처

https://sparkbyexamples.com/spark/spark-how-to-create-an-empty-dataset/

Spark - 비어있는 Dataset을 만드는 방법

이 글에서, 저는 몇가지 Scala 예시를 사용하여 비어있는 Spark Dataset(emptyDataset())을 스키마가 있고 없고에 따라 만드는 방법을 설명하겠습니다. 우리는 시작하기 전에, 비어있는 Dataset을 만들 필요가 있는 많은 시나리오 중에 하나를 설명하겠습니다.

Spark에서 파일로 작업하는 동안 처리할 파일을 받지 못하는 경우도 있지만, 파일을 받을 때 생성한 데이터 세트와 유사한 (동일한 스키마) 빈 데이터 세트를 생성해야 합니다. 동일한 스키마로 생성하지 않으면 표시되지 않을 수 있는 열을 참조하므로 데이터 세트에 대한 작업 / 변환이 실패합니다.

관련글: Spark 비어있는 DataFrame 생성하기

이와 유사한 상황을 처리하려면 항상 동일한 스키마로 Dataset을 생성해야 합니다. 즉, 파일이 존재하거나 빈 파일 처리에 관계없이 동일한 열 이름과 데이터 유형을 의미합니다.

먼저 예제 전체에서 사용할 SparkSession 및 Spark StructType 스키마와 case class를 생성해 보겠습니다.

val spark:SparkSession = SparkSession.builder()
   .master("local[1]")
   .appName("SparkByExamples.com")
   .getOrCreate()

import spark.implicits._

val schema = StructType(
    StructField("firstName", StringType, true) ::
      StructField("lastName", IntegerType, false) ::
      StructField("middleName", IntegerType, false) :: Nil)

val colSeq = Seq("firstName","lastName","middleName")
case class Name(firstName: String, lastName: String, middleName:String)

emptyDataset() – 컬럼이 없는 비어있는 Dataset 생성

SparkSession은 스키마 없이 빈 Dataset을 반환하는 emptyDataset() 메서드를 제공하지만 이것은 우리가 원하는 것이 아닙니다. 다음 예제는 스키마로 생성하는 방법을 보여줍니다.

case class Empty()
val ds0 = spark.emptyDataset[Empty]
ds0.printSchema()
// Outputs following
root

emptyDataset() – 스키마로 비어있는 Dataset 생성

아래 예에서는 스키마 (열 이름 및 데이터 type)가 있는 Spark 빈 데이터 세트를 만듭니다.

val ds1=spark.emptyDataset[Name]
ds1.printSchema()
// Outputs following
root
 |-- firstName: string (nullable = true)
 |-- lastName: string (nullable = true)
 |-- middleName: string (nullable = true)

createDataset() – 스키마로 비어있는 Dataset 생성

SparkSession의 createDataset() 메서드를 사용하여 스키마가 있는 빈 Spark Dataset을 만들 수 있습니다. 아래의 두 번째 예는 먼저 빈 RDD를 생성하고 RDD를 데이터 셋으로 변환하는 방법을 설명합니다.

val ds2=spark.createDataset(Seq.empty[Name])
ds2.printSchema()
val ds3=spark.createDataset(spark.sparkContext.emptyRDD[Name])
ds3.printSchema()
//These both Outputs following
root
 |-- firstName: string (nullable = true)
 |-- lastName: string (nullable = true)
 |-- middleName: string (nullable = true)

createDataset () – 기본 열 이름으로 빈 Dataset 만들기

val ds4=spark.createDataset(Seq.empty[(String,String,String)])
ds4.printSchema()
// Outputs following
root
 |-- _1: string (nullable = true)
 |-- _2: string (nullable = true)
 |-- _3: string (nullable = true)

암시적인 인코더 사용

암시적인 인코더를 사용하는 다른 방법을 살펴 보겠습니다.

val ds5 = Seq.empty[(String,String,String)].toDS()
ds5.printSchema()
// Outputs following
root
 |-- _1: string (nullable = true)
 |-- _2: string (nullable = true)
 |-- _3: string (nullable = true)

case class 사용

Scala case class에서 원하는 스키마로 빈 데이터 세트를 만들 수도 있습니다.

val ds6 = Seq.empty[Name].toDS()
ds6.printSchema()
// Outputs following
root
 |-- firstName: string (nullable = true)
 |-- lastName: string (nullable = true)
 |-- middleName: string (nullable = true)

출처
https://stackoverflow.com/questions/37791685/what-is-the-use-case-for-start-awaittermination-and-stop-with-regard-to-sp

스파크 스트리밍에 관해 start(), awaitTermination(), stop()의 사용 사례는 무엇입니까?

저는 스파크 스트리밍 초보입니다. 저는 터미널에서 데이터를 추출하여 HDFS로 불러오는 하나의 응용프로그램을 개발하고 있습니다. 인터넷에서 찾아 보았지만 스트리밍 응용프로그램을 멈추는 방법을 이해할 수 없었습니다.

또한 sc.awaittermination()과 sc.stop()의 사용 사례를 저에게 설명해 주실 수 있으신가요?

감사합니다.

2개의 답변 중 1개의 답변만 추려냄

streamingContext.awaitTermination() --> 사용자로부터 종료 신호를 기다립니다. 사용자로부터 신호를 받을 때(예시 CTRL+c 또는 SIGTERM) 스트리밍 context는 멈출 것입니다. 이는 java의 shutdownhook 종류입니다.

streamingContext.stop() --> 스트리밍 context를 바로 멈춥니다. 스파크 context에 관해 스트리밍 context에 말할 수 있습니다. 만약 스파크 context가 아니고 스트리밍 context만 멈추기를 원한다면 streamingContext.stop(false)를 호출할 수 있습니다.

출처

https://stackoverflow.com/questions/7938585/what-does-param-mean-in-scala

스칼라에서 파라미터:_* 는 무엇을 뜻합니까?

스칼라(2.9.1)로 새로워지면서, List[Event]가 있고 이를 Queue[Event]로 복사하고 싶습니다. 하지만 다음 문법은 대신에 Queue[List[Event]]를 yield 합니다. 

val eventQueue = Queue(events)

이러한 이유로 다음은 작동합니다.

val eventQueue = Queue(events : _*)

하지만 저는 이것이 무엇인지 왜 작동 하는지 이해하고 싶습니다. Queue.apply 함수의 signature를 이미 보았습니다.

def apply[A](elems: A*)

그리고 저는 첫 번째 시도가 왜 작동하지 않는지와 두 번째(_*)의 의미가 무엇인지 이해하고 싶습니다. 이 경우 :와 _*는 무엇입니까? 그리고 왜 apply함수는 Iterable[A]를 취하지 않는 것입니까?

이 발생할 것입니다.

1개의 답변만 발췌

a: A는 타입 선언(ascription)입니다. 스칼라에서 타입 선언(ascriptions)의 목적은 무엇입니까? 라는 글을 보세요.

: _*는 가변 인자 시퀀스로 시퀀스 타입을 하나의 인수로 취급 하겠다고 컴파일러에게 알리는 타입 선언(ascription)의 특별한 인스턴스 입니다.

시퀀스나 iterable한 하나의 요소를 가지는 Queue.apply를 사용한 Queue를 생성하는 것은 완전히 유효합니다. 그래서 이는 하나의 Iterable[A]가 주어질 때 정확히 발생합니다.

출처

https://stackoverflow.com/questions/30674376/apache-spark-get-elements-of-row-by-name

Apache Spark: Row의 element값을 이름으로 얻기

Apache Spark(저는 Scala 인터페이스를 사용합니다)의 DataFrame 객체에서, 객체를 순회한다고 했을 때 이름으로 값들을 추출하는 방법이 있을까요? 저는 정말 어색한 방법으로 이 작업을 했습니다.

def foo(r: Row) = {
  val ix = (0 until r.schema.length).map( i => r.schema(i).name -> i).toMap
  val field1 = r.getString(ix("field1"))
  val field2 = r.getLong(ix("field2"))
  ...
}
dataframe.map(foo)

저는 더 좋은 방법이 있을 것이라 생각합니다. 제 작업은 매우 장황한 방법입니다. 이는 추가 구조를 만들어야 하고 명시적으로 type도 알아야 합니다. 만약 타입이 틀리면 프로그램은 compile-time 오류보다 실행시 exception이 발생할 것입니다.

1개의 답변만 발췌

당신은 org.apache.spark.sql.Row로부터 "getAs"를 사용할 수 있습니다.

r.getAs("field1")
r.getAs("field2")

getAs(java.lang.String fieldName) 관해 링크를 클릭하시면 더 아실 수 있습니다.

출처

https://stackoverflow.com/questions/44450889/why-does-spark-shell-fail-to-load-a-file-with-class-with-rdd-imported/44451056

왜 spark-shell은 RDD를 import한 class를 가진 파일을 불러오기를 실패할까요?

저는 Scala 2.11.8로 Spark 2.1.1을 사용합니다.

spark-shell에서 저는 RDD를 메소드로 가지는 클래스를 불러오기 위해 :load명령을 사용합니다.

그 클래스를 불러올 때 컴파일 오류가 나옵니다.

error: not found: type RDD

왜일까요? import 구문이 있습니다.

다음은 제가 작업한 코드입니다.

답변

spark-shell에서 :load의 특징인듯 합니다. 해결책은 당신의 클래스 정의하는 부분으로 import org.apache.spark.rdd.RDD(.이나 _ 없이)를 이동하는 것입니다.

이는 RDD클래스에만 국한 된 것이 아니고 모든 클래스에 해당합니다. import문이 클래스 자체 내에 정의되어 있지 않으면 작동하지 않습니다.

말했듯이, 다음은 클래스 밖에 import하고 있기 때문에 작동하지 않을 것입니다.

import org.apache.spark.rdd.RDD
class Hello {
  def get(rdd: RDD[String]): RDD[String] = rdd
}

scala> :load hello.scala
Loading hello.scala...
import org.apache.spark.rdd.RDD
<console>:12: error: not found: type RDD
         def get(rdd: RDD[String]): RDD[String] = rdd
                                    ^
<console>:12: error: not found: type RDD
         def get(rdd: RDD[String]): RDD[String] = rdd

:load의 -v플래그를 사용하여 무슨 일이 발생하는 지 볼 수 있습니다.

scala> :load -v hello.scala
Loading hello.scala...

scala>

scala> import org.apache.spark.rdd.RDD
import org.apache.spark.rdd.RDD

scala> class Hello {
     |   def get(rdd: RDD[String]): RDD[String] = rdd
     | }
<console>:12: error: not found: type RDD
         def get(rdd: RDD[String]): RDD[String] = rdd
                                    ^
<console>:12: error: not found: type RDD
         def get(rdd: RDD[String]): RDD[String] = rdd
                      ^

이는 클래스 정의 안에서 import를 하는 것이 도움이 될 것이라고 생각하게 되었습니다. 그리고 (저는 크게 놀라며) 실제로 그랬습니다!

class Hello {
  import org.apache.spark.rdd.RDD
  def get(rdd: RDD[String]): RDD[String] = rdd
}

scala> :load -v hello.scala
Loading hello.scala...

scala> class Hello {
     |   import org.apache.spark.rdd.RDD
     |   def get(rdd: RDD[String]): RDD[String] = rdd
     | }
defined class Hello

또한 :paste명령을 사용하여 클래스를 spark-shell에 붙여 넣을 수 있습니다. 고유한 패키지에 클래스를 정의할 수 있는 raw 모드가 있습니다.

package mypackage

class Hello {
  import org.apache.spark.rdd.RDD
  def get(rdd: RDD[String]): RDD[String] = rdd
}

scala> :load -v hello.scala
Loading hello.scala...

scala> package mypackage
<console>:1: error: illegal start of definition
package mypackage
^

scala>

scala> class Hello {
     |   import org.apache.spark.rdd.RDD
     |   def get(rdd: RDD[String]): RDD[String] = rdd
     | }
defined class Hello

scala> :paste -raw
// Entering paste mode (ctrl-D to finish)

package mypackage

class Hello {
  import org.apache.spark.rdd.RDD
  def get(rdd: RDD[String]): RDD[String] = rdd
}

// Exiting paste mode, now interpreting.

// 붙여넣기 모드를 종료하고 인터프리터 모드입니다.

출처 : https://stackoverflow.com/questions/37201918/spark-job-keep-on-running

Spark 작업을 실행한 채 유지하기

저는 다음 명령을 사용하여 ambari-server에서 저의 Spark 작업을 제출 하였습니다.

./spark-submit --class  customer.core.classname --master yarn --numexecutors 2 --driver-memory 2g --executor-memory 2g --executor-cores 1 /home/hdfs/Test/classname-0.0.1-SNAPSHOT-SNAPSHOT.jar newdata host:6667

이 명령은 잘 작동하였습니다.

하지만, 명령 프롬프트를 닫거나 그 작업을 kill하려고 할 때 그 작업은 실행한 채 유지하여야 합니다.

어떠한 도움이든 감사합니다.

4개 답변 중 1개의 답변

몇 가지 방법으로 이를 이룰 수 있습니다.

1) nohup을 사용하여 백그라운드(background)로 드라이버(driver) 프로세스를 spark-submit으로 실행할 수 있습니다.

nohup  ./spark-submit --class  customer.core.classname \
  --master yarn --numexecutors 2 \
  --driver-memory 2g --executor-memory 2g --executor-cores 1 \
  /home/hdfs/Test/classname-0.0.1-SNAPSHOT-SNAPSHOT.jar \
  newdata host:6667 &

2) 드라이버 프로세스가 다른 노드에서 실행되도록 배포 모드에서 클러스터로 실행하십시오.