Pyspark 사용 예시(2)

view는 메모리상의 가상 table과 같은 구조이고 실제로 존재하진 않는다.

Spark는 메모리에 view라는 테이블을 만들고 그 view를 sql로 다룬다.

 

from pyspark.sql import SparkSession

sparkSession = SparkSession.builder.appName("pyspark-sql-test").getOrCreate()
df = sparkSession.read.csv('hdfs://localhost:9000/user/diabetes_info/diabetes.csv')
df.show()

df.createOrReplaceTempView("diabetes")
df2 = sparkSession.sql("select count(*) from diabetes")

df3 = sparkSession.sql("select avg(_c6) from diabetes")

df2.show()
df3.show()

Output

1. df.createOrReplaceTempView("diabetes") 

-> 데이터프레임(df)의 내용으로 "diabetes"라는 view를 생성한다

 

2. df2 = sparkSession.sql("select count(*) from diabetes")

-> 이 문장에서 사용된 'diabetes'는 1번에서 만든 view의 이름이다

-> 그 view를 대상으로 sql문을 작성한 것이다. (실행해보면 MapReduce 대비 100배 빠름이 느껴진다)

 

---

from pyspark.sql import SparkSession

sparkSession = SparkSession.builder.appName("pyspark-hdfs1").getOrCreate()

#hdfs 영역으로부터 파일 읽어오기
#df = sparkSession.read.csv('hdfs://localhost:9000/user/csv/sample.csv')

# 로컬에서 파일 로드
# employee.csv 파일에는 컬럼명이 포함되있음
#예시 1)
df1 = sparkSession.read.csv('/home/hdoop/employee.csv')  # header를 설정 안해준 경우
df1.show()

#예시 2)
df2 = sparkSession.read.csv('/home/hdoop/employee.csv', header=True)
df2.show()

#employee_copy.csv 파일에는 컬럼명이 없음
df3 = sparkSession.read.csv('/home/hdoop/employee_copy.csv')
df3 = df2.toDF('id','name','salary','job')
df3.show() 

df4 = sparkSession.read.csv('/home/hdoop/employee_copy.csv')
column_name = ['id','name','salary','job']
df4 = df2.toDF('id','name','salary','job')
df4.show()

Output

1. sparkSession.read.csv()

-> csv 파일의 내용을 가져올 수 있다.

-> 예시 1) 과 2)의 차이는 괄호() 안에 header의 유무인데 header=True를 사용하면

    csv파일의 컬럼명을 데이터프레임의 컬럼명으로 사용할 수 있다

 

2. df.toDF(컬럼명)을 통해 header=true 조건으로 처음부터 컬렴명을 가져오지 않고

먼저 데이터프레임을 생성하고 나서 따로 컬럼명을 지정해줄 수 있다.

*df.toDF() 안에는 literable한 데이터가 들어가면 된다.