[파이썬] Hadoop과 Spark를 이용한 빅데이터 처리

1. Hadoop

 

1-1. Hadoop이란

 

 

데이터를 다루다 보면 용량이 매우 큰 데이터를 다뤄야 할 때가 있습니다.

이것을 컴퓨터 하나에서만 처리하기에는 무리가 있습니다.

 

따라서 매우 큰 용량의 데이터를 처리하기 위한 도구가 필요했습니다.

 

이를 위해 나온게 Hadoop입니다.

 

Hadoop은 데이터를 분산 저장, 분산 처리하는 플랫폼으로

상용화된 일반적인 컴퓨터 클러스트( 일반적인 집에서 사용하는 컴퓨터 )에서

매우 큰 용량의 데이터를 다루기 위해 개발되었습니다.

 

Hadoop은 다음과 같은 특징을 가지고 있습니다.

• Distributed storage: 하나의 저장소에 국한되지 않고 클러스터에 컴퓨터를 추가하면 해당 컴퓨터의 저장장치가 데이터 저장소의 일부가 됨
• Distributed processing: 클러스터에 속한 컴퓨터들이 함께 데이터를 처리함( 병렬 처리 )

 

Hadoop의 장점은 아래와 같습니다.

• 저렴한 비용 -> 상용화된 컴퓨터로 가능
• 안전함 -> 분산 저장
• 빠름 -> 분산 처리
• 확장성 -> 상용화된 컴퓨터
• 다양한 데이터 유형 지원
• 다양한 도구 및 프레임워크 지원

 

Hadoop의 단점은 아래와 같습니다.

• 작은 크기의 파일은 비효율적: 네트워크로 통신해야 하기 때문에 작은 파일은 로컬로 하는게 나음
• 높은 지연시간: 실시간 응답에 약함
• 복잡한 설치 및 운영
• concept write once, read many: 수정/삭제가 까다롭고 성능이 낮음
• 보안 설정이 까다로움: 하드웨어 공유로 성능을 높였기 때문에 데이터 접근이 쉬움

 

1-2. Hadoop HDFS

 

 

HDFS는 Hadoop Distributed File System이며

대용량 파일을 분산된 서버에 저장하고, 많은 클라이언트가 저장된 데이터를

빠르게 저장할 수 있게 설계된 파일 시스템입니다.

 

네트워크 기반 파일 시스템이며 빅데이터를 Hadoop 클러스터 컴퓨터들에

분산 저장하고 클러스터의 하드 드라이브를 하나의 거대한 파일 시스템처럼 사용합니다.

 

1-3. Hadoop Map Reduce

 

 

분산 처리를 위한 필수 처리방식으로 Mapper와 Reducer로 구성됩니다.

Hadoop에 내재된 기능이며 클러스터에 데이터의 처리를 분배합니다.

 

Mapper는 클러스터에 분산된 데이터를 효율적으로 동시에 변형해서

최종적으로 원하는 결과를 고려해 데이터를 변형하는 단계입니다.

 

Reducer는 이렇게 변형된 데이터를 집계합니다.

 

2. Spark

 

2-1. Spark란

 

 

Hadoop은 위 사진과 같은 생태계를 가지고 있습니다.

 

 

이 중 Spark는 Map Reduce의 아이디어를 계승해 실행하는 엔진을 새롭게 설계했습니다.

 

클러스터 환경에서 데이터를 병렬로 처리하는 라이브러리 집합으로

데이터를 신속하고 효율적이며 안정적으로 처리합니다.

 

In-Memory연산을 통해 디스크 기반 Hadoop의 성능을 향상하며

RDD, DataFrame, Dataset 3가지의 API를 제공합니다.

 

Spark의 구성은 다음과 같습니다.

• Driver program: Task 진행을 통제하며 SparkContext를 생성하는 메인 process
• Executor: 다수의 worker node에서 실행되는 프로세스로 Caching을 통해 빠르고 효율적으로 데이터를 가공
• Cluster manager: Executor에 task를 할당하고 관리

 

SparkContext는 Spark 기능의 기본 진입점으로 클러스터와 통신을 담당합니다.

 

Spark는 파이썬, 자바, 스칼라, R을 지원하며 Spark SQL이라는 인터페이스로 SQL query또는 유사함수를 통해

Spark의 data를 변환합니다.

 

728x90

 

2-2. Spark 설치

 

https://www.oracle.com/java/technologies/javase/jdk17-archive-downloads.html

Java Archive Downloads - Java SE 17.0.12 and earlier

 

 

먼저 Java를 설치해 줍니다.

 

Java는 17버전 SDK를 설치해야 합니다.

 

 

 

윈도우 버전을 설치해 줍시다.

 

Java를 설치했다면 Anaconda를 실행하여 다음 명령어를 입력합니다.

 

pip install pyspark == 3.5.5

 

pyspark 3.5.5버전을 설치하는 명령어입니다.

 

다음 컴퓨터에 Hadoop을 구성하기 위해 Hadoop을 설치합니다.

 

https://github.com/cdarlint/winutils/tree/master/hadoop-3.3.6/bin

winutils/hadoop-3.3.6/bin at master · cdarlint/winutils

 

위 링크에서 Hadoop 3.3.x버전의 winutils.exe와 Hadoop.dll 파일을 다운로드 합니다.

 

 

C드라이브에서 Hadoop\bin을 생성 후, 위 두 파일을 저장합니다.

 

다음 자바와 Hadoop의 환경 변수를 설정합니다.

 

Java의 Path는 Java가 설치된 폴더입니다.

 

 

System변수와 사용자 변수 Path에 모두 추가합니다

 

2-3. Python에서 Spark 사용

 

data.zip

2.46MB

 

위 데이터를 다운받고 C 드라이브 밑에 datas 폴더를 만들어 그 폴더에 압축 해제해 줍니다.

 

Jupyter Notebook으로 새로운 ipynb파일을 하나 만들어줍니다.

 

다음 아래 명령어를 입력해 줍니다.

 

import pyspark
from pyspark.sql import SparkSession

path = 'C:\\datas\\students.csv'

 

경로를 설정할 때 C:\\\a\\\b\\\c\\\d 와 같은 형식으로 설정해야 합니다.

 

다음 Spark Session을 하나 만들어 줍니다.

 

spark = SparkSession.builder.appName('01').config('spark.driver.host','localhost').getOrCreate()

 

SparkSession.builder는 SparkSession을 생성하는 명령어입니다.

appName은 aplication의 이름을 설정하는 것입니다.

config('spark.driver.host','localhost')는 aplication의 host를 local로 설정합니다.

getOrCreate는 기존 SparkSession을 가져오거나 새롭게 생성합니다.

 

이제 데이터를 Read해 줍니다.

SparkSession의 Read는 데이터를 로드한 후, DataFrame 타입으로 반환합니다.

 

students = spark.read.csv(path)

students.show(5)

 

 

우리는 데이터가 클러스터 환경에서 분산되어 저장된 큰 용량의 데이터로 가정합니다.

 

따라서 지금 현재 driver가 실행 중인 host에 모든 데이터를 가져와서 확인하는 것은

불가능하다고 간주합니다.

 

DataFrame 전체 출력은 불가하며 show를 통해 일부만 확인할 수 있습니다.

 

DataFrame.show(n)은 n개의 행만 반환한다는 뜻입니다.

 

students.printSchema() # column이 자동 생성되면 _c0,_c1... 등으로 채워짐 nullable은 null값의 유무

 

DataFrame의 Schema를 확인합니다.

Column 명과 data type, nullable 정보를 확인할 수 있습니다.

 

그런데 column이 모두 _c0, _c1, _c2...처럼 되어있는 모습을 확인할 수 있습니다.

 

첫 행이 column으로 인식되지 못해 생기는 문제입니다.

 

또한 모든 데이터 타입이 string으로 되어있습니다.

 

이는 read.csv에 header, inferSchema와 같은 파라미터 설정을 통해 해결할 수 있습니다.

 

students = spark.read.csv(path,header = True,inferSchema=True)
students.show(5)
students.printSchema()

 

header는 True로 설정할 경우, 첫 행을 column명으로 간주합니다.

 

inferSchema는 Treu로 설정할 경우, 각 column의 type과 nullable을 확인합니다.

대용량 데이터의 경우, True로 설정하면 read 속도가 매우 느려질 수 있으므로 

직접 schema를 입력하는 것이 좋습니다.

 

이번에는 json파일을 read합니다.

 

path_json = 'C:\\datas\\people.json'
people = spark.read.json(path_json)
print(type(people))
people.show(3)
people.printSchema()

 

json은 header를 설정하지 않아도 column명이 자동으로 할당됩니다.

 

Schema를 지정하기 위해 다음 명령어를 입력합니다.

 

from pyspark.sql.types import StructField, StructType, IntegerType, StringType, DateType

# time은 DateType, 나머지 colunm은 StringType으로 지정
people_field = [
    StructField('name',StringType(),True),
    StructField('city',StringType(),True),
    StructField('timestamp',DateType(),True),
    StructField('creditcard',StringType(),True)
]

 

StructField는 한개의 column( name, data type, nullable )을 정의합니다.

StructType은 여러 개의 StructField가 모인 list 형태의 container ( Schema )입니다.

 

이제 StructType생성해 schema를 설정합니다.

 

# StructField를 원소로하는 list로 StructType 생성
people_schema = StructType(fields=people_field)

people2 = spark.read.json(path_json,schema=people_schema)

people.printSchema()
people2.printSchema()

 

DataFrame을 저장하기 위해선 다음과 같은 명령어를 사용하시면 됩니다.

 

people2.write.mode('overwrite').csv('C:\\datas\\people')

 

mode는 쓰기 모드를 설정하며, csv(path)로 저장할 path를 지정합니다.