빅데이터 부트캠프 56일차
Spark 기본(DataFrame)
- spark.createDataFrame(데이터, 스키마) = 해당 데이터로 해당 스키마에 맞춰서 데이터프레임 생성 (상세하게)
- 시퀀스 데이터.toDF(컬럼명,컬럼명,..) = 해당 시퀀스 데이터로 스키마 없어도 알아서 데이터 프레임 생성 (편하게)
(사용하는 시퀀스 데이터의 타입을 보통 seq,list로 사용하나 rbb도 가능)
- StructType() = 스키마 객체 생성 (괄호안에 StructField를 이용해 각 열에 대한 속성 지정)
- StructField(컬럼명, 데이터 타입, nullable 사용여부) -> 데이터프레임 컬럼 설정
- StructType(array(StructField, StructField)) 형식 말고도 new StructType().add().add() 형식으로 스키마 생성 가능
- spark.read.csv() = csv파일 불러오기
- spark.read.text() = txt파일 불러오기
- spark.read.json() = json파일 불러오기
- 하둡에서 스파크로 데이터 불러올 때는 리눅스 경로와는 다른 형태로 입력
[hdfs://하둡 노드주소 및 포트번호/하둡에 있는 가져올 파일 경로]
ex) spark.read.csv("hdfs://hadoop-name:9000/Spark_Dataset/simple_zipcodes.csv")
- spark.sql() = ()안에 sql문을 입력하여 sql문을 사용 가능하게 해줌
여기서부터는 기본적으로 [데이터프레임명.] 형식으로 입력
- column.slice() = 인덱스 슬라이싱 적용하여 출력
- withColumn(컬럼명, 데이터) =
컬럼명이 기존에 있는 컬럼명인 경우에는 기존 컬럼 데이터 수정
컬럼명이 기존에 없는 컬럼명일 경우에는 새로운 컬럼을 추가하여 그 안에 지정한 데이터 입력
(lit이나 기존의 컬럼에 있는 데이터를 이용하여 데이터를 입력해줘야함, 컬럼명만 입력하면 Error)
- withColumn(컬럼명, lit(텍스트)) = 새로운 컬럼을 추가하여 그 안에 데이터로 지정한 텍스트가 반복되어 입력되도록 실행
- withColumn(컬럼명, col(기존컬럼명) 수식) = 기존컬럼에 있는 데이터에 수식을 적용한 새 데이터를 적용
- withColumn(컬럼명, col(기존컬럼명.중첩컬럼명)) = 중첩컬럼을 해체하여 지정한 컬럼명의 새로운 컬럼으로 생성
- withColumn(컬럼명, col(기존컬럼명).cast(데이터타입)) = 해당 컬럼에 있는 데이터들의 타입을 지정한 타입으로 변경
- withColumnRenamed('바꿀 컬럼명', '바꾼 컬럼명') = 지정한 컬럼명 변경
- 여러개의 컬럼명을 변경하려면 withColumnRenamed().withColumnRenamed() 처럼 이어서 작성하여 실행
- createOrReplaceTempView('객체명') = sql문으로 데이터프레임을 조작할 수 있도록 데이터프레임을 지정한 객체명으로 객체화
- drop('컬럼명') = 해당 컬럼 삭제
- dropDuplicates() = 지정한 컬럼 기준(여러개 가능)으로 데이터가 중복되는 데이터들 첫번째 데이터 제외하고 모두 삭제
- show() = 데이터프레임 출력 (숫자를 지정하여 출력될 행의 개수 설정 가능)
- select() = 해당 데이터프레임에서 특정 컬럼만 추출하고 싶을 때 사용
- colRegex() = 정규표현식 사용하여 조건 입력
- filter() = where() = 조건을 만족하는 데이터 반환
- when() = 조건을 만족하는 데이터들을 어떤 방식으로 반환할지 지정후 반환
- otherwise() = when을 만족하지 않는 데이터들을 어떤 방식으로 반환할지 지정후 반환
- withColumn("컬럼명", when(), otherwise()) =
컬럼명이 기존에 있는 컬럼명인 경우에는 기존 컬럼 데이터를 지정한 조건에 맞게 수정
컬럼명이 기존에 없는 컬럼명일 경우에는 새로운 컬럼을 추가하여 그 안에 지정한 조건에 맞게 데이터 입력
withColumn("컬럼명", when(), otherwise()) 대신에 select(col("*"), when(), otherwiste()) 사용 가능
※ 조건문 입력시 '컬럼명, col("컬럼명"), df("컬럼명") 는 입력방식이 다를뿐 모두 특정 컬럼을 지정하는 같은 기능
※ '컬럼명, col("컬럼명"), df("컬럼명")을 이용하여 조건 입력시 ==이 아닌 === 사용 (!=는 !==으로 사용)
※ ', col, df를 이용하지 않고 " "안에 조건문 입력시에는 == 사용
- 조건문을 when(), otherwise() 대신에 expr("case when" + "when" + "else") 사용가능
expr("case when ~ then ~" + "when ~ then ~" + "else ~") "을 이미 사용하고 있어 문자열 텍스트 입력시에는 '사용
- distinct("컬럼명") = 중복 데이터 제거
- groupby("컬럼명").통계함수 = 해당 컬럼을 기준으로 데이터 그룹화한 후 지정한 통계함수를 적용한 값을 새로운 칼럼을 생성하여 반환
- groupby().pivot().통계함수 = 그룹화 기준 컬럼을 행으로 전환하고 나머지 행 원소들을 열로 전환후 통계함수로 지정한 값을 출력하여 데이터프레임을 반환
※ 통계함수
- sum(), count(), min(), max(), avg(), mean()
- 통계함수 안에는 여러개의 컬럼명을 입력할 수 있고 각각의 통계함수 적용 값이 반환
- 다양한 통계함수를 적용하고 싶으면 agg(sum(), avg())를 이어서 입력하여 적용 가능
(agg안에 있는 통계함수 뒤에 .as()를 이어서 입력하면 새로 생기는 컬럼의 이름을 지정할 수 있음)
- agg()안의 통계함수가 아닌 일반 통계에 .as()를 이어서 입력해도 컬럼 이름 지정 불가능
(컬럼명 변경하려면 withColumnRenamed 이용해야함)
ex) groupby('part').sum('salary').as('A')
- sort("컬럼명") = OrderBy("컬럼명") = 데이터 정렬
- sort나 OrderBy에 컬럼명 입력시 그냥 "컬럼명", $"컬럼명", col("컬럼명") 형태 모두 가능
- sort()안에 있는 $"컬럼명", col("컬럼명")뒤에 .asc를 붙이면 오름차순, .desc를 붙이면 내림차순으로 정렬
(그냥 "컬럼명"은 .asc와 .desc 적용시 에러)
- join()
- union("또 다른 데이터프레임명") = 동일한 스키마 구조를 가진 두 데이터프레임을 하나로 합침
(중복값 제외시켜서 합치고 싶다면 뒤에 .distinct() 입력)
- map() = 데이터프레임의 각 행 데이터에 지정한 함수 적용후 결과값 반환
- flatmap() = 데이터프레임의 모든 행 데이터를 하나로 합쳐 함수 적용후 결과값 반환