Pyspark 사용 예시(1)

기본적으로 Hadoop과 연동해서 사용하는 것이므로 hadoop이 작동중이어야 한다. (start-all.sh)

 

import findspark
findspark.init()
from pyspark.sql import SparkSession

sparkSession = SparkSession.builder.appName("example-pyspark-hdfs").getOrCreate()
df_load = sparkSession.read.csv('hdfs://localhost:9000/user/diabetes_info/diabetes.csv')

# appName 확인
sparkSession.sparkContext.appName   # example-pyspark-hdfs

print( type(df_load) )
df_load.show()              # 화면에 표시할 때 사용
print( dir(df_load) )

df_load.write.csv('hdfs://localhost:9000/user/data/csv/spark')  # 이미 존재하면 오류발생
print('CSV copy saved...')

Output

주피터 노트북에 작성한 예시이다.

*실행 시 에러가 발생하는 경우도 있는데 vmware 환경에서 돌리고 있다면 vmware를 껏다가 키면 된다...(나는 그랬음)

 

위에서 작성한 sparkSession.read.csv(hdfs://localhost: 로 시작하는 부분은 사전에 다른 작업하다가 만들어둔 hdfs영역의 csv파일이다. 본인이 테스트 하고자 하는 hdfs영역의 파일 경로를 저기에 작성하면 된다.

밑에 write.csv('hdfs  로 시작하는 부분은 spark가 작성할 파일의 경로와 파일 이름이다. 원하는 경로를 지정해주면 된다.

 

 

Spark 프로그램의 특징은 사용하기 전에 항상 SparkSession 객체를 생성해야 한다.

1. sparkSession = SparkSession.builder.appName("example-pyspark-hdfs").getOrCreate()

-> SparkSession의 객체를 생성한 것이다. appName()은 쓰고싶지 않다면 작성하지 않아도 된다. 기본적으로 괄호() 안의 이름을 세션의 이름으로 지정하는 것이다. 이미 동일한 이름으로 존재한다면 그 세션을 참조한다. getOrCreate()를 통해 세션을 새로 만들거나 가져오는 것이다.

 

2. df_load = sparkSession.read.csv() 

-> sparkSession 객체 안의 read라는 함수로 괄호()안의 데이터를 읽어온다.

-> localhost:9000은 사전에 지정해준 번호이다.

-> 위 예시에서는 괄호에 'hdfs'로 시작하는데 이 뜻은 spark가 hdfs 영역의 내용을 읽는다는 것이다.

-> 객체명이 df로 시작되는데 이 뜻은 read.csv를 이용해 읽어온 데이터를 DataFrame으로 저장된다는 것을 의미한다.

     - 이 DataFrame은 Scala의 DataFrame이고 Pandas의 DataFrame과는 다른 객체이다.

     - 그래도 Spark의 DataFrame과 Pandas의 DataFrame을 호환되도록 하는 Tool이 존재한다.

 

3. df_load.write.csv  

-> df_load 데이터를 어딘가에 작성한다는 의미(데이터프레임은 메모리에 저장이 되고 그 저장된 값을 write을 통해 작성한다)

-> 디렉토리의 마지막 경로 까지만 입력해야한다. 파일명까지 작성하는 것이 아님. (무작위로 이름이 지어진다. e.g. part-0000) 

-> 그 경로에 파일이 이미 존재하면 에러가 발생한다(이미 있다면 덮어 쓰도록 하는 명령어도 있음)

 

4. sparkSession.sparkContext.appName

-> sparkSession 객체를 만들 때 지정했던 세션의 이름이 나온다

 

5. df_load.show()

-> 스파크의 데이터프레임을 출력할 때는 .show()를 이용한다 (20개를 출력한다)

 

6. dir(df_load)

-> df_load 데이터프레임의 구조와 변수, 함수등이 아주 상세하게 출력된다

 

7. import findspark,  findspark.init()

spark 라이브러리가 이미 설치가 되어있고 pyspark을 설치하면 환경변수 우선순위로 인해(?) pyspark를 사용하지 못하는 경우가 있는데 그런 경우 findspark를 이용하면 가능하다고 한다.

설치할 때 환경변수를 잘 적용하면 저 부분이 없어도 아무런 문제가 없이 돌아간다.