출처 : https://stackoverflow.com/questions/34629313/how-to-measure-the-execution-time-of-a-query-on-spark

Spark에서 쿼리의 실행 시간을 측정하는 방법

저는 Apache spark (Bluemix)에서 쿼리의 실행 시간을 측정할 필요가 있습니다. 제가 시도한 내용입니다.

import time

startTimeQuery = time.clock()
df = sqlContext.sql(query)
df.show()
endTimeQuery = time.clock()
runTimeQuery = endTimeQuery - startTimeQuery

좋은 방법입니까? 제가 얻은 시간은 테이블에서 보았을 때 상대적으로 너무 작은 거 같습니다.

9 개의 답변 중 1 개의 답변

spark-shell(Scala) 에서 이를 하기 위해서, spark.time()를 사용할 수 있습니다..

저의 다른 답변도 보실 수 있습니다 : https://stackoverflow.com/questions/36389019/spark-query-execution-time/50289329#50289329

df = sqlContext.sql(query)
spark.time(df.show())

출력은 다음처럼 나올 것입니다.

+----+----+
|col1|col2|
+----+----+
|val1|val2|
+----+----+
Time taken: xxx ms

관련 글 : 성능 문제 해결을 위한 Apache Spark 워크로드 메트릭 측정.

출처

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/24718697/pyspark-drop-rows

PySpark에서 행 버리기

PySpark에서 RDD로부터 행을 어떻게 버릴 수 있을까요? 특별히 첫 번째 행에 제 데이터 셋에 컬럼명이 포함되어 있기 때문입니다. API를 자세히 살펴보면 이 작업을 쉽게 하는 방법을 찾을 수 없는 거 같습니다. 당연히 저는 Bash / HDFS를 통해 이를 할 수 있지만 PySpark로만 이를 할 수 있는 방법을 알고 싶습니다.

6개 답변 중 1개만 추려냄

제가 아는 한 이를 하는 '쉬운' 방법은 없습니다.

그래서 트릭을 수행해야 합니다.

val header = data.first
val rows = data.filter(line => line != header)

출처
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/50791975/spark-structured-streaming-multiple-writestreams-to-same-sink

같은 Sink로 여러개의 Spark Structured Streaming WriteStreams하기

Spark Structured Streaming 2.2.1에서 순서대로 같은 데이터베이스 sink로 두 개의 Writestream 하는 것이 안 됩니다. 이 2개의 Writestream이 순서대로 실행하는 방법을 제안해주세요.

val deleteSink = ds1.writestream
  .outputMode("update")
  .foreach(mydbsink)
  .start()

val UpsertSink = ds2.writestream
  .outputMode("update")
  .foreach(mydbsink)
  .start()

deleteSink.awaitTermination()
UpsertSink.awaitTermination()

위의 코드에서 사용한대로 deleteSink는 UpsertSink 뒤에 실행됩니다.

1개의 답변

만약 당신이 병렬로 두개의 stream을 실행하고 싶으시면 다음을 사용해야 합니다.

sparkSession.streams.awaitAnyTermination()

다음 것 대신에 말이지요.

deleteSink.awaitTermination()
UpsertSink.awaitTermination()

당신 코드의 UpsertSink의 경우 deleteSink가 멈추거나 exception이 발생하지 않으면 시작하지 않을 것입니다. scaladoc에 이런 내용이 나와 있습니다.

exception 발생 또는 query.stop() 또는 this 쿼리의 종료를 기다립니다. exception과 함께 쿼리가 종료되면 exception이 발생될 것입니다. 만약 쿼리가 정상 종료되면, 이 메소드와 모든 후속 호출은 바로 리턴될 것입니다. (쿼리가 stop()에 의해 종료되면) exception이 바로 발생할 것입니다. (쿼리가 exception이 발생하여 종료했다면)