Spark Sql 실습

흠.. 어제 하둡설정한게 뭔가 이상한갑.. 갑자기 스팤이 안된다. 그래서 hadoop설정파일을 일단 지우고 sparksql을 해보려한다.

SparkSession

엔트리 포인트가 되는 스파크 세션을 생성한 후 데이터를 읽어보자

예제를 똑같이 따라해서 데이터 프레임의 내용을 보니 나이와 이름이 적혀있다. 실제로 이 데이터 프레임은 아래에 존재한다.

스파크설치경로/examples/src/main/resources/people.json

쿼리문 실행해보기

아무런 문제 없이 잘된다.ㅎㅎ 이 외에도 쿼리문을 직접 날려서 사용할 수 있다.

사용하기전 createOrReplaceTempView를 이용하여 SQL임시보기로 등록해야한다.

하지만 위와 같은 임시보기는 세션이 종료되면 사라진다. spark application이 종료될때까지 유지시키려면 전역 임시보기를 만들 수 있다.

데이터 셋 생성

위의 사진처럼 class의 형태로 데이터 셋의 형태를 생성하고 json파일을 읽어서 매핑시켜서 화면에 보이게 할 수 있다.

그리고 중간의 DS는 인코더인데 코드를 동적으로 생성하고...spark에서 바이트를 개체로 역 직렬화하지 않고 필터링,정렬,해싱과 같은 많은 작업을 수행 할 수 있는 형식을 사용

인코더가 잘이해가안되는데..음

그냥 보여주려는건가...

만약 이 둘(json파일과 클래스)의 컬럼 길이가 불일치하면 아래와 같이 오류가 뜬다

RDD와 DataFrame

sparksql은 RDD를 DataFrame으로 변환시킬 수 있다.

두가지 방법이 있는데

1.리플렉션을 사용하여 특정 유형의 객체를 포함하는 RDD 스키마 추정

2.스키마를 작성한 다음 기존 RDD에 적용할 수 있는 프로그래밍 인터페이스 사용

일단 그냥 봐선 1도 모르겠다. 실습을 해보자

1.리플렉션을 사용하여 특정 유형의 객체를 포함하는 RDD 스키마 추정        *리플렉션이 뭐지?   "객체를 통해 클래스의 정보를 분석해 내는 프로그래밍 기법"

공식문서 "Class에 대한 인수의 이름은 리플렉션을 사용하여 읽고 열의 이름이 됩니다"

*위에 코드에서 보면  .map이 있는데 찾아보니 데이터를 가공한다고 한다.

자 일단 예제 코드를 모두 따라해봤지만 오류 없이 잘 돌아갔다.(오류 넘무섭 ㅠ_ㅠ) 여기서 중요한 부분처럼 보이는것은

val peopleDF = spark.sparkContext
  .textFile("examples/src/main/resources/people.txt")
  .map(_.split(","))
  .map(attributes => Person(attributes(0), attributes(1).trim.toInt))
  .toDF()

..이 부분은 txt파일로부터 Person 오브젝트로 RDD를 생성해서  DataFrame으로 변환하는 것이다.

2.스키마를 작성한 다음 기존 RDD에 적용할 수 있는 프로그래밍 인터페이스 사용

*노란색 부분은 오류가 나서 해결한 부분

자 1번과의 차이점을 알아보자

txt파일을 읽어와서 RDD로 만드는것은 똑같은 것 같다.

val schemaString = "name age"

val fields = schemaString.split(" ")
  .map(fieldName => StructField(fieldName, StringType, nullable = true))
val schema = StructType(fields)


val rowRDD = peopleRDD
  .map(_.split(","))
  .map(attributes => Row(attributes(0), attributes(1).trim))


val peopleDF = spark.createDataFrame(rowRDD, schema)

이 코드가 다른 것 같다. 스키마를 생성하는 것!