ShaShaKa Dev Note

scalatest를 이용하여, spark app을 테스트하는 예제를 작성해보도록 하겠습니다. 기본 템플릿은 이전 create first app을 참고 부탁드립니다. 기존에 만들어진 FirstApp을 기반으로, scalatest를 작성해보도록 합니다. 가장 먼저 test실행시 사용할 BaseSpec을 작성해주도록 합니다. - BaseSpec.scala package org.shashaka.test

가장 간단한 형식에서, 기본 데이터 소스(parquet)는 모든 작업에 사용됩니다. Dataset<Row> usersDF = spark.read().load("examples/src/main/resources/users.parquet"); usersDF.select("name", "favorite_color").write().save("namesAndFavColors.parquet"); Manually Specifying Options 데이터 소스에 전달하려는 추가 옵션과 함께 사용될 데이터 소스를 수동으로 지정할 수도 있습니다. 데이터 소스는 완전한 이름(예 : org.apache.spark.sql.parquet)으로 지정되지만 내장 소스의 경우 짧은 이름 (json, parquet, jdbc, orc, libsvm, csv, text)을 사용할 수도 있습니다. 모든 데이터 소스 유형에서 로드된 DataFrame은 syntax를 사용하여 다른 유형으로 변환될 수 있습니다. JSON 파일을 로드하려면 다음을 사용할 수 있습니다. Dataset<Row> peopleDFCsv = spark.read().format("csv") .option("sep", ";") .option("inferSchema", "true") .option("header", "true") .load("examples/src/main/resources/people.csv"); 추가 옵션은 쓰기 작업 중에도 사용됩니다. 예를 들어 ORC 데이터 소스의 블룸 필터 및 사전 인코딩을...

Spark SQL의 임시 뷰는 세션 범위이며 이를 생성하는 세션이 종료되면 사라집니다. 모든 세션에서 공유되는 임시 뷰를 원하고 Spark 애플리케이션이 종료 될 때까지 활성 상태로 유지하려면 global temporary view를 작성할 수 있습니다. Global temporary view는 데이터베이스 global_temp에 연결되어 있으며 정규화된 이름을 사용하여 참조해야합니다 (예 : select * FROM global_temp.view1.) // Register the DataFrame as a global temporary view df.createGlobalTempView("people");

SparkSession의 sql 함수를 사용하면 응용 프로그램에서 프로그래밍 방식으로 SQL 쿼리를 실행하고 결과를 Dataset<Row>로 반환 할 수 있습니다. import org.apache.spark.sql.Dataset; import org.apache.spark.sql.Row;

Components

다음 표는 Spark에서 지원하는 몇 가지 일반적인 작업을 보여줍니다. RDD API 문서 (Scala, Java, Python, R)를 참조하십시오. 자세한 내용은 RDD 함수 doc (Scala, Java)을 참조하십시오. * reduce(func) 함수 func 두 개의 인수를 사용하여 하나를 반환)를 사용하여 데이터 집합의 요소를 집계합니다. 함수는 병렬로 올바르게 계산될 수 있도록 정류적이고 연관성이 있어야합니다. * collect() 데이터 세트의 모든 요소를 ​​드라이버 프로그램에 배열로 반환합니다. 이것은 일반적으로 데이터의 작은 하위 집합을 반환하는 필터 또는 다른 작업 후에 유용합니다. * count() 데이터 세트의 전체 요소의 갯수를 반환합니다. * first() 데이터 세트의 가장 처음 요소를 반환합니다. take(1) 메서드 호출과 비슷한 기능입니다. * take(n) 데이터 세트의 처음 n개 요소가있는...

Spark 내의 특정 작업은 셔플이라고하는 이벤트를 트리거합니다. 셔플 (shuffle)은 데이터를 재분배하여 Spark간에 다르게 그룹화되는 Spark의 메커니즘입니다. 이것은 일반적으로 executor와 machine간에 데이터를 복사하고, 복잡하고 값비싼 작업인 shuffle을 만드는 것을 포함하고 있습니다.

대부분의 Spark 작업은 모든 유형의 객체가 포함된 RDD에서 작동하지만 일부 특수한 작업은 키-값 쌍의 RDD에서만 사용할 수 있습니다. 가장 일반적인 예는 요소를 key를 기준으로 그룹화 또는 집계하는 것과 같은 분산된 "suffle"작업입니다. Java에서 키-값 쌍은 Scala 표준 라이브러리의 scala.Tuple2 클래스를 사용하여 표시됩니다. new Tuple2(a, b)를 호출하여 튜플을 만들고 나중에 tuple._1 () 및 tuple._2 ()를 사용하여 해당 필드에 액세스 할 수 있습니다. 키-값 쌍의 RDD는 JavaPairRDD 클래스로 표시됩니다. mapToPair 및 flatMapToPair와 같은 특수한 경우의 map 작업을 사용하여 JavaRDD로부터 JavaPairRDD를 만들 수 있습니다. 예를 들어, 다음 코드는 키-값 쌍에서 reduceByKey 연산을 사용하여 파일에서 각 텍스트 행이 몇 번 발생하는지 계산합니다.

Spark의 가장 어려운 점 중 하나는 클러스터에서 코드를 실행할 때 변수 및 메서드의 범위와 수명주기를 이해하는 것입니다. 범위 밖의 변수를 수정하는 RDD 작업은 빈번한 혼동의 원인이 될 수 있습니다. 아래 예에서는 foreach ()를 사용하여 카운터를 늘리는 코드를 살펴 보지만 다른 작업에서도 비슷한 문제가 발생할 수 있습니다.

Spark는 로컬 파일 시스템, HDFS, Cassandra, HBase, Amazon S3 등 Hadoop이 지원하는 모든 스토리지 소스에서 분산 데이터 세트를 생성 할 수 있습니다. Spark는 텍스트 파일, SequenceFile 및 기타 Hadoop InputFormat을 지원합니다. 텍스트 파일 RDD는 SparkContext의 textFile 메서드를 사용하여 만들 수 있습니다. 이 메소드는 파일의 URI (시스템의 로컬 경로 또는 hdfs : //, s3a : // 등의 URI)를 가져 와서 각각의 행으로 이루어진 collection처럼 읽습니다. 다음은 호출 예제입니다.