from pyspark import SparkConf, SparkContext

# spark setting
conf = SparkConf().setMaster("local").setAppName("MinTemperatures")
sc = SparkContext(conf = conf)


# 텍스트 파일로 읽어들인 RDD를 가공하기 쉽게 (key, value1, value2)로 가공하는 함수
def parseLine(line):
    # csv seperate ','
    fields = line.split(',')
    stationID = fields[0]
    entryType = fields[2]
    # C -> F
    temperature = float(fields[3]) * 0.1 * (9.0 / 5.0) + 32.0
    return (stationID, entryType, temperature)

# csv 파일을 텍스트 RDD로 생성
lines = sc.textFile("1800.csv")
# (key, **value)로 변환
parsedLines = lines.map(parseLine)
# 최대 온도를 구하기 위해 TMAX Type만을 필터링
minTemps = parsedLines.filter(lambda x: "TMAX" in x[1])
# 집계 함수를 활용하기 위해 (key, value)형식으로 변환
stationTemps = minTemps.map(lambda x: (x[0], x[2]))
# reduceByKey 집계 함수를 사용하여 key값 별로 최대 값만 남김
minTemps = stationTemps.reduceByKey(lambda x, y: max(x,y))
# 모든 RDD값을 반환
results = minTemps.collect();

for result in results:
    print(result[0] + "\t{:.2f}F".format(result[1]))

1. SparkConf와 SparkContext를 생성합니다.

from pyspark import SparkConf, SparkContext
import collections

conf = SparkConf().setMaster("local").setAppName("RatingsHistogram")
sc = SparkContext(conf=conf)

2. 데이터를 읽어 들입니다.

lines = sc.textFile("./ml-100k/u.data")

3. 평점을 추출합니다.

ratings = lines.map(lambda x: x.split()[2])

4. 평점을 세어 봅니다.

result = ratings.countByValue()

5. 결과를 정렬합니다.

sortedResults = collections.OrderedDict(sorted(result.items()))

6. 결과를 출력합니다.

for key, value in sortedResults.items():
    print("%s %i" % (key, value))

결론

Spark는 RDD를 사용하여 데이터를 처리합니다. RDD는 분산된 데이터 세트이며, Spark는 RDD를 사용하여 병렬로 작업을 수행할 수 있습니다. ```

Spark의 핵심은 Resilient Distributed Dataset (RDD) -> 복구 가능한 분산 데이터 셋

RDD

• 다양한 데이터 세트를 추상화 한 것.

  RDD는 분산되고 변형하는 성질을 갖고 있어 여러 클러스터에 나눌 수 있고 개인 컴퓨터에서도 작동 가능. 클러스터의 특정 노드에 이상이 생겨도 자동으로 처리, 노드 하나가 작동을 멈춰도 계속 작동하면서 작업을 재분배.

• RDD는 큰 데이터 셋 -> 한 데이터를 다른 데이터로 변환하는데 쓰임

Spark context -> sc

sc객체를 사용해서 RDD를 생성 할 수 있음.

RDD 변형

• map -> 데이터를 불러와 다른 데이터로 변형. 기존 RDD의 모든 엔트리가 새 RDD의 새로운 값으로 변형

  rdd.map(lamdba x: x**x

• flatmap -> map과 비슷한 기능을 수행, 단 새로 생성된 RDD가 기존 RDD보다 크거나 작을 수 있음.

• filter -> RDD에 있는 데이터에서 일부분만 보고 싶을 때 사용

• distinct -> RDD에서 중복값 제거 함수

• uninon, intersaction, subtract -> RDBMS와 동일

RDD 집계

• collect -> 원하는 값만 출력

• countByValue -> 특정한 값이 얼마나 나왔는가

• take, top -> 최종적으로 완성한 RDD에서 일부 값을 가져온다

• reduce -> 모든 값에 특정한 연산을 해서 하나의 RDD로 합쳐준다

유향 비순환 그래프

Spark가 빠른이유. Spark에 액션을 입력하면 바로 유향 비순환 그래프를 생성하여 원하는 값을 얻으려면어떻게 해야하는지 최적화된 방법을 계산.

Pig를 사용하면 더 쉽게 데이터를 처리할 수 있다.

Hadoop과 MapReduce 위에 구축되었기 때문에 이를 통해 매퍼와 리듀서를 작성하지 않고 MapReduce 작업을 할 수 있다.

• Apache Pig-> 매퍼나 리듀서를 작성하지 않고 데이터를 빠르게 분석할 수 있는 방법

MapReduce 개발 사이클 타임

MapReduce의 가장 큰 문제는 개발 사이클 타임

MapReduce 프로그램을 개발해 실행하고 원하는 작업을 수행하기까지 오랜 시간이 걸린다.

그래서 Pig가 등장

Pig는 MapReduce 위에서 작동하고 Pig Latin이라는 새 프로그래밍 언어를 사용 -> SQL과 비슷하다.

다른 점은 Pig는 절차형 언어라는 겁니다 아주 간단한 스크립트 양식을 사용해 단계별로 데이터 간에 여러 관계를 설정.

MapReducer in Hadoop

정말 큰 데이터 세트를 가진 클러스터를 운영하고 있다고 가정하면, 처리 과정을 여러 컴퓨터에 배분하거나 적어도 여러 작업(task)에 걸쳐 진행해야만 한다.

같은 컴퓨터 안에서 다른 프로세스나 컨테이너를 실행할 수도 있다는 것. 혹은 컴퓨터 여러 대에 걸쳐 분배했을 수도 있다 예를들어 3개의 노드에 나눠 매핑한다고 가정한다면, 데이터를 가져오며 세 부분으로 나눠 각각의 노드에 분배함

분산처리가 가능한 이유

• 매핑 단계에서는 다른 노드 있는 줄을 신경쓰지 않아도 됨으로 작업을 병렬화하기에 수월.
• 입력 데이터 한 덩어리를 한 컴퓨터에서 매핑하고 다른 덩어리는 또 다른 컴퓨터에서 동시에 매핑한다면 Hadoop은 이 작업이 끝나면 정보를 받아오기만 하면 됨.

셔플과 정렬 과정에서 분산처리는?

셔플과 정렬 과정을 여러 노드로 나눠 작업을 하게 된다면, 다른 노드에서 같은 키 값이 나올 수도 있다. 이를 같은 키끼리 모아 리듀서로 보내야하는데 어떻게 하는걸까

예를 들어 ID=1의 사용자의 값이 두 노드에 걸쳐있고 셔플과 정렬 작업에서 이 둘은 함께 집계되게 되는데 MapReduce가 이 작업을 대신 수행한다. 네트워크를 활용하여 여러 노드들간 데이터를 보내는 방식을 사용하지 않음. 대신 이 모든 데이터를 merge sort하게 된다.

즉 각각의 노드에서 맵퍼를 통해 키-값 구조로 구조화 시킨 후 합병 정렬을 통해 셔츨과 정렬 과정을 수행한다.

MapReducer Architecture

• 클라이언트 노드가 작업을 개시 -> MapReduce 작업을 지시합
• 클라이언트는 먼저 YARN(job)과 소통 -> 어떤 MapReduce 작업이 필요한지

  YARN -> 리소스 교섭자, 어떤 머신에서 무엇을 실행할지 관리. 클러스터의 어떤 머신이 가용하고 어떤 머신의 성능은 얼마인지 등의 정보를 기억

• 데이터를 HDFS나 적절한 분산 파일 시스템에 복사합니다
• 'MapReduce Master'

  MapReduce Master는 개별 매핑과 리듀싱 작업을 주시하고 리소스 관리자와 협업해 작업을 클러스터에 걸쳐 배분합니다

MapReduce는 HDFS, YARN과 함께 제공되는 Hadoop의 핵심 기술.

• Hadoop에 내재된 기능이며 클러스터에 데이터의 처리를 분배.
• MapReduce는 데이터를 파티션으로 나눠서 클러스터에 걸쳐 병렬 처리되도록 한다

Data Mapper

매핑은 기본적으로 데이터를 변형시킨다 즉 데이터가 한 줄씩 들어오면 매퍼는 그 데이터를 변형시킨다.

• 키-값 쌍으로 전환

예를 들어 각 사용자가 시청한 영화의 수를 집계하려고 한다면, 키는 사용자의 ID, 값은 사용자가 평가한 영화가 될 것. 일반적인 python dict 자료구조와는 달리 중복된 키가 존재할 수 있음. 이 경우 ID=1의 사용자가 두개 이상의 영화를 평가한 것. 맵핑 단계에서 같은 키값이 나오는 것은 전혀 문제될 것이 없음.

들어오는 데이터에서 필요한 정보를 추출하고 이해할 수 있는 구조로 정리. 모든 입력 줄(input line)마다 매퍼는 중요한 데이터를 추출하고 구조화해한 줄을 출력한다.

'셔플과 정렬'

'셔플과 정렬' 단계에서 일어나는 일은 각각의 키-값 쌍을 읽어서 고유 키에 짝 지어진 값을 집계. 간단하게 여러개의 중복된 키가 존재한다면 이를 하나로 묶어줌.

Date Reducer

Reducer의 역할은 데이터를 집계하는 것.

정리

• 매퍼는 각 줄을 키-값 쌍으로 변형

• '셔플과 정렬'은 각 데이터를 집계해 정렬

• 리듀서는 이 데이터를 원하는 대로 처리할 수 있죠

이전에 포스팅했던 새로운 방식으로 코드를 변경했다.

@api.get('/moveline')
def get_moveline(request, moves: str = None):
    board = chess.Board()
    if moves == None:
        data = list(ChessNotation.objects.annotate(submoves=Substr('mainline', 1, 4)).exclude(mainline='').values('submoves').annotate(cnt=Count('submoves'), white_win=Count(
            'result', filter=Q(result__iexact='white')), draw=Count('result', filter=Q(result__iexact='draw')), black_win=Count('result', filter=Q(result__iexact='black'))).order_by('-cnt'))

        return JsonResponse(data, safe=False)
    else:
        move = moves.split(',')
        data = list(ChessNotation.objects.filter(mainline__startswith=moves).exclude(mainline=moves).annotate(submoves=Substr(
            'mainline', 1+5*len(move), 4)).values('submoves').annotate(cnt=Count('submoves'), white_win=Count(
                'result', filter=Q(result__iexact='white')), draw=Count('result', filter=Q(result__iexact='draw')), black_win=Count('result', filter=Q(result__iexact='black'))).order_by('-cnt'))
        return JsonResponse(data, safe=False)

orm이 기존에비해 많이 길어지긴 했지만, 속도측면이나 데이터베이스 용량측면에서 보다 나아졌다고 확신한다.

다중 annotate 조건

django annotate를 사용할때 조건을 적용할 수 있다.

ChessNotation.objects.filter(mainline__startswith=moves).exclude(mainline=moves).annotate(submoves=Substr(
            'mainline', 1+5*len(move), 4)).values('submoves').annotate(cnt=Count('submoves'), white_win=Count(
                'result', filter=Q(result__iexact='white')), draw=Count('result', filter=Q(result__iexact='draw')), black_win=Count('result', filter=Q(result__iexact='black'))).order_by('-cnt'))

프로젝트를 진행하면서 생각나는대로 즉흥적으로 api설계를 하고 모델들을 정의하다보니 전반적으로 진행이 매끄럽지 않은듯한 느낌이 들었다.

이미 진행했던 프로젝트를 리팩토링 하는 겸 요구사항을 명확하게 정의하여 보다 깔끔한 코드를 작성하기 위해 요구사항을 정리해보려한다.

요구사항

• lichess database에서 제공하는 pgn파일을 처리할 수 있는 데이터 파이프라인 구축

• 체스 분석 기능

  • 체스 시작 포지션 부터 실제 게임에서 나온 수들을 카운트해서 보여줄 수 있어야함.
  • stockfish를 활용하여 특정 포지션에서의 최선의 수를 찾는다.
    • stockfish를 활용하여 현재 포지션의 유불리를 판단.
    • 레이팅에 따른 다른 수들을 보여줘야함. (1800~2000 의 수들만 보여준다든가)

• 체스 퍼즐

  • 체스 퍼즐을 테마에 따라 분류하는 기능
    • 체스 퍼즐을 풀면서 실제로 수를 움직이기전에 미리둬보는 기능

• stockfish와 배틀

  • stockfish의 난이도를 직접 설정할 수 있어야함.

• Dashboard

  • lichess database에서 가져온 데이터를 활용하여 여러가지 분석된 결과를 제공해야함 (특정 레이팅에서 가장 승률이 높은 오프닝, 가장 많이 사용된 오프닝, 첫수에 따른 오프닝 선택, 전체 오프닝의 승률 ..etc)

lichess database에서 추출한 74만개의 체스 기보 데이터를 가지고 lichess나 chess.com에서 제공하는 analze기능을 따라잡아보자

체스는 전략과 계획이 필요한 대표적인 보드 게임. 체스에서 성공하기 위해서는 다양한 기술과 전략을 학습하고 연습해야 한다. 실력을 향상시키기 위해서는 자신의 게임을 분석하고, 상대방의 전략을 이해하는 것이 매우 중요하다.

lichess나 chess.com과 같은 사이트들에서 여러가지 퍼즐이나 여러가지 상황에 따른 다음 수들을 실제 행해졌던 게임들에서 수집하여 보여주는 기능들이 있고, 여러가지 체스 오프닝들이나, 전략을 공부할 수 있는 퍼즐과 같은 기능들을 제공한다.

평소 체스에 관심이 많았던 사람으로써, 개인 프로젝트로 lichess나 chess.com사이트의 기능들을 클론하고, 내가 학습하기 용이한 기능들을 추가하여, 개인적인 체스 분석, 연습 사이트를 만드는 프로젝트를 진행하고 있다.

나만의 체스 분석 프로그램을 만들어보자 프로젝트

이번 포스팅은 lichess 분석기능에 대해서 정리한다.

기존 방법

기존의 방법은 lichess database에서 제공하는 pgn파일들을 가져와 데이터를 수집하고 정제하여 데이터베이스에 적재하는 파이프라인을 구축하였다.

파이프라인 코드 예시

class Processor:

    def __init__(self, fpath, num, checkpoint, is_start=False) -> None:
        self.df = pd.DataFrame()
        self.num = num
        self.msg = 'done'
        self.fpath = fpath
        self.checkpoint = checkpoint
        self.status = False
        self.next_checkpoint = checkpoint + num
        self.is_start = False

        b, m = self.parse_data_to_csv()
        if b:
            self.msg = self.store_data()
        else:
            self.msg = m

    def parse_data_to_csv(self):
        pgn = open(self.fpath, 'r')
        if self.is_start:
            for _ in range(1):
                game = chess.pgn.read_game(pgn)

        for i in range(self.checkpoint):
            game = chess.pgn.read_game(pgn)

        game = chess.pgn.read_game(pgn)
        num = 0
        if game:
            while game != None:
                if num == self.num:
                    break
                if not self.df.empty:
                    df = pd.DataFrame(
                        list(game.headers.values()) + [self.san_to_uci(game.mainline_moves())]).T
                    df.columns = list(game.headers.keys()) + ['mainline']
                    self.df = pd.concat([self.df, df], ignore_index=True)
                else:
                    self.df = pd.DataFrame(
                        list(game.headers.values()) + [self.san_to_uci(game.mainline_moves())]).T
                    self.df.columns = list(game.headers.keys()) + ['mainline']

                num += 1
                game = chess.pgn.read_game(pgn)

            if game == None:
                self.status = True
                self.df.drop(self.df.index[-1])

            return True, 'done'
        else:
            return False, f'{self.fpath} is already done.'

            .......

lichess database에서 제공하는 파일들은 기본적으로 크기가 매우 크기 때문에 파일을 분산하여 처리하는 api를 만들어 데이터베이스에 저장했다.

여기서 저장한 데이터를 활용하여, 체스의 id라고 할 수 있는 fen을 활용하여 각각의 fen마다 사람들이 둔 수들을 하나씩 세서 데이터베이스에 저장했다.

@api.get('/mainline')
def parse_mainline(request, num: int):
    pool = Pool(2)
    first_pk = ChessNotation.objects.first()
    try:
        checkpoint = ChessMainline.objects.all()[0]
        if checkpoint.checkpoint < first_pk.pk:
            checkpoint.checkpoint = first_pk.pk
    except:
        checkpoint = ChessMainline.objects.create(checkpoint=first_pk.pk)
    data = ChessNotation.objects.filter(
        pk__gte=checkpoint.checkpoint, pk__lt=checkpoint.checkpoint + num)

    checkpoint.checkpoint += num

    checkpoint.save()
    if data:
        # MainlineProcessor(data)

        pool.map(MainlineProcessor, data)

    return

class MainlineProcessor:
    def __init__(self, data: ChessNotation) -> None:
        self.data = data
        self.df = pd.DataFrame()
        self.current = None

        self.parse()

    def process(self, next_move, result, fen):
        try:
            start = ChessProcess.objects.get(fen=fen)
        except:
            try:
                start = ChessProcess.objects.create(fen=fen)
            except:
                return
        try:
            created_move = start.next_moves.get(next_move=next_move)
        except:
            created_move = ChessFenNextMoves.objects.create(
                fen=start, white=0, draw=0, black=0, next_move=next_move, cnt=0)

        created_move.cnt += 1
        if result == 'white':
            created_move.white += 1
        elif result == 'black':
            created_move.black += 1
        else:
            created_move.draw += 1

        created_move.save()

    def parse(self):
        i = self.data
        # for i in self.data:
        board = chess.Board()
        moves = i.mainline.split(',')

        self.process(moves[0], i.result, board.fen())

        for cnt in range(len(moves) - 1):
            board.push_uci(moves[cnt])
            fen = board.fen()

            self.process(moves[cnt+1], i.result, fen)

python의 내장 라이브러리인 multiprocessing을 활용했음에도 불구하고 100개의 데이터를 정제하는데 대략 30초의 시간이 소요됐고, 각각의 fen을 따로 데이터베이스에 저장함에 따라 100000개의 기보를 정제하는데 3000000만개의 fen 데이터가 생성되는 아주 비효율적인 방법을 사용하였다.

수정된 방법

수정된 방법은 아주 간단하고 빠르다. django orm기능을 적극 활용한다.

import chess

board = chess.Board()
legal_moves = list(board.legal_moves)

for move in legal_moves:
    print(<database model name>.objects.filter(mainline__startswith = move).count(), move)

정말 간단하다.. 머리가 나쁘면 몸이 고생한다

lichess database 기준 대략 74만개의 게임을 분석.

하위 레이팅 1200 ~ 1800, 중위 레이팅 1900 ~ 2400, 상위 레이팅 2500~ 로 설정하고 각 레이팅 별로 어떤 오프닝이 가장 인기있는지 알아보자.

하위 레이팅

most_count

most_win_white

most_draw

most_win_black

중위 레이팅

most_count

most_win_white

most_win_draw

most_win_black

상위 레이팅

most_count

most_win_white

most_win_draw

most_win_black

전체 레이팅

most_count

most_win_white

most_win_draw

most_win_black

이전 포스팅에서는 lichess에서 제공한 체스 오프닝 데이터를 간단하게 전처리 했다. 이번엔 본격적으로 우리가 데이터베이스에 저장할 포맷으로 전처리를 진행한다.

데이터 구조

데이터를 전처리하기전에 어떤식으로 구조를 짜야 쿼리를 쉽고 빠르게 할 수 있을지에 대해서 고민을 해보다 2가지 선택지를 생각했다.

• double linked list
• hash table

결론부터 말하자면 hash table (dict)의 구조를 채용하여 전처리 하기로 하였다.

double linked list의 구조도 역시 활용할만 하였으나 여러 예외처리들과 오프닝 데이터를 쿼리를 할 때 특정 오프닝의 이름을 알기 위해서 끝까지 탐색해야 한다는 점이 비효율적이라고 생각했다.

체스 기보를 나타내는 방식엔 pgn방식과 uci방식이 있는데 후에 stockfish를 활용하기 위하여 stockfish와 호환이 잘되는 uci방식으로 데이터를 저장한다.

스톡피시는 꽤 오랜 기간동안 사용된 대중에게 공개된 가장 강력한 체스 엔진입니다. 무료 오픈 소스 엔진인 스톡피시는 현재 사람들이 자발적으로 참여하여 개발되고 있습니다. 스톡피시는 2004년 토르드 롬스타드(Tord Romstad)가 개발한 엔진을 기반으로 하며, 이후 마르코 코스탈바(Marco Costalba)가 2008년 이를 발전시켰습니다. 주나 키이스키(Joona Kiiski)와 게리 린스콧(Gary Linscott)또한 창립자로 여겨집니다. 출처 (https://www.chess.com/ko/terms/stockfish-chess-engine-ko#what)

#pipelines.py
import pandas as pd
from .utils import _revmove_number


FPATH = "<path>"
FNAME = 'pre_data.csv'

class PreProcessing:
    def __init__(self, fpath) -> None:
        self.fname = FNAME
        self.fpath = FPATH
        self.__fpath = fpath
        self.raw_data = None
        self.data_preprocessing()

    def data_preprocessing(self):
        df =self.import_data_csv(fpath=self.__fpath)
        self.raw_data = df
        df.to_csv(f'{self.fpath}{self.fname}')


    def import_data_csv(self, fpath, sep = '\t'):
      data = pd.read_csv(fpath, sep=sep)
      df = self.split_data(data)

      df = pd.DataFrame(df, columns=['pgn','uci','name','wdl'])

      return df


    def split_data(self, data):
      df = []
      for i in data.index:
        val = data.loc[i, 'uci'].split(' ')
        pgn_val = data.loc[i, 'pgn'].split(' ')
        k = _revmove_number(pgn_val)
        name = data.loc[i, 'name']
        df.append([k,val,name,''])

      df.sort(key=lambda x: len(x[0]))

      return df

#process.py
import pandas as pd
from ast import literal_eval
import csv
DEFAULT_FILE_NAME = 'preprocessing_opening.csv'
FPATH = "C:\\Users\\aaa57\\chess\\dist\\"

class Hash:
  def __init__(self) -> None:
    self.default_dic = {
          'next' : [],
          'name' : 'default',
        }

class ProcessData:
  def __init__(self,data) -> None:
    self.data = data
    self.hash_data = {
      'start' : Hash().default_dic
    }

    for i in self.data.index:
      try:
        self.to_hash(literal_eval(self.data.loc[i,'uci']), self.data.loc[i,'name'])
      except:
        self.to_hash(self.data.loc[i,'uci'], self.data.loc[i,'name'])

    self.to_csv()

  def to_hash(self, uci, name):
    if len(uci) == 1:
      self.hash_data[str(uci)] = Hash().default_dic
      self.hash_data['start']['next'].append(uci[-1])
      self.hash_data[str(uci)]['name'] = name

    else:
      try:
        self.hash_data[str(uci[:-1])]['next'].append(uci[-1])
      except:
        self.to_hash(uci[:-1],name)
        self.hash_data[str(uci[:-1])]['next'].append(uci[-1])

      self.hash_data[str(uci)] = Hash().default_dic
      self.hash_data[str(uci)]['name'] = name

  def to_csv(self):
    with open(f'{FPATH}{DEFAULT_FILE_NAME}', "w", newline="", encoding='utf-8') as csv_file:
    # Create a CSV writer object
      writer = csv.writer(csv_file)

      # Write the header row
      writer.writerow(["Move", "Next Moves", "Opening Name"])

      # Iterate over the dictionary and write each row to the CSV file
      for move, details in self.hash_data.items():
        # Convert the string representation of the list back to a list
        if move != 'start':
          move = eval(move)
        next_moves = ",".join(details["next"])
        writer.writerow([move, next_moves, details["name"]])


class ProcessUtil:

  def to_hash(self, row):
    if len(row['pgn']) == 1:
      return

데이터가 생성되면 pipelines.py 에서 1차적으로 데이터를 추출 변형시키고 process.py에서 1차적으로 처리된 데이터를 데이터베이스에 저장할 수 있도록 hash table형태로 저장한다.

#Processor.py
from .pipelines import PreProcessing
from process.process import ProcessData


class DataProcessor:
  def __init__(self):
    self.data = None

    self.data_processor("C:\\Users\\aaa57\\chess\\dist\\chess_opening.csv")
    self.process()

  def data_processor(self, fpath):
    self.data = PreProcessing(fpath=fpath).raw_data

  def process(self):
    ProcessData(self.data)

후에 지속적으로 체스 기보 데이터가 들어올 때마다 코드를 일일히 실행하는 것은 비효율적이기 때문에 DataProcessor라는 객체를 생성해 코드를 최소한으로 작성해 실행할 수 있도록 하였다.

이번 프로젝트의 구성은 Stockfish를 활용하여 Chess engine과의 대결을 할 수 있는 사이트와, chess 기보를 읽어들이는 data pipeline, 그 데이터를 저장하는 data warehouse를 구성하는 것이 목표.

Why Chess?

평소 체스와 바둑 장기 등 여러 보드게임을 즐겨하는 중에 체스에 대한 매력을 느껴 여러 사이트를 전전하던 중 사람들과 대결한 기보를 체스 엔진에다가 분석을 맡기는 기능을 유료로 서비스하는 사이트들이 많았다.

물론 lichess같이 무료로 모든 것을 제공하는 사이트도 있지만, datapipeline을 구축해보고 엄청난 data(약 1.2TB)를 다뤄볼 수 있다는게 엄청난 매력 아닌가.

이번 프로젝트를 통해 대용량 데이터를 관계형 데이터베이스(MYSQL, POSTGRESSQL)과 NoSQL(MongoDB, HBASE)를 활용해서 대용량 데이터베이스에서 빠른 쿼리 응답을 수행할 수 있도록 하는 것으로 목표를 잡았다. (이 데이터를 활용해서 여러 기능들을 추가하는것은 덤.)

Chess Data

모든 체스 데이터는 lichess.data에서 가져왔음. 오픈소스고 무료!

이번 포스트는 체스 오프닝 데이터를 처리하고 데이터베이스에 저장하는 포스팅을 해보겠다.

lichess에서 제공해주는 체스 오프닝 데이터는 5개의 tsv파일로 이루어져 있고

dataframe으로 나타내면 이렇다.

import pandas as pd
data = pd.read_csv(fpath, sep="\t")

print(data)

데이터 수집을 용이하게 하기 위해 5개의 tsv파일을 합쳐주자

data = pd.read_csv(fpath, sep="\t")

k = ['b','c','d','e']

for i in k:
  fpath = <filepath>
  sub = pd.read_csv(fpath, sep='\t')

  data = pd.concat([data, sub])

data.to_csv(<path>/<to>/<store>, sep='\t')

pandas를 활용하여 간단하게 tsv파일을 합칠 수 있다.

Chess opening Pipeline

기본적으로 데이터를 저장하는 구조는 business process model의 구조를 활용 할 생각이다.

사진 출처 (https://www.researchgate.net/figure/Petri-net-model-that-captures-flexible-business-process-scenario_fig2_220830118)

비지니스 프로세스 모델을 한줄로 설명하면 시작에서 완료까지 계획된 비즈니스 프로세스의 단계를 모델링하는 흐름도 메소드. lichess에서 제공하는 오프닝 데이터의 기보를 살펴보면, 체스는 검은색, 흰색 플레이어가 번갈아가면서 한 수씩 두는 게임으로

1. 뒤에 Nh3는 흰색이 d5은 검은색이 둔 수이다. 비지니스 프로세스 모델처럼 시작과 끝이 있고 순차적인 데이터인걸 확인 할 수 있다. 한 줄의 데이터만 본다면 위의 프로세스 모델처럼 될 수 없지만, 여러개의 데이터를 조합하면 위의 모델을 도출할 수 있다.

Feature of chess opening data

위의 사진은 체스 오프닝 데이터의 일부이다.

['1.', 'Nh3', 'd5', '2.', 'g3', 'e5', '3.', 'f4', 'Bxh3', '4.', 'Bxh3', 'exf4'] Amar Gambit ['1.', 'Nh3'] Amar Opening ['1.', 'Nh3', 'd5', '2.', 'g3', 'e5', '3.', 'f4', 'Bxh3', '4.', 'Bxh3', 'exf4', '5.', 'O-O', 'fxg3', '6.', 'hxg3'] Amar Opening: Gent Gambit ['1.', 'Nh3', 'd5', '2.', 'g3', 'e5', '3.', 'f4'] Amar Opening: Paris Gambit ['1.', 'e3', 'e5', '2.', 'c4', 'd6', '3.', 'Nc3', 'Nc6', '4.', 'b3', 'Nf6'] Amsterdam Attack ['1.', 'a3'] Anderssen's Opening ['1.', 'a3', 'a5', '2.', 'b4'] Anderssen's Opening: Polish Gambit

데이터들의 한가지 공통점은 특정 부분을 기점으로 오프닝의 이름이 바뀐다는 것.

• ['1.', 'Nh3'] Amar Opening

Nh3로 시작되는 오프닝은 Amar Opening이라는 이름이 붙고 위의 예시 데이터에서 Nh3로 시작되는 오프닝은

Amar Gambit, Amar Opening: Gent Gambit, Amar Opening: Paris Gambit이 존재한다. 그리고 이들 간 f4이후 분기가 나뉘고 Bxh3이후엔 완전히 나눠진다.

Nh3로 시작되는 다른 이름이 오프닝도 있지만 위의 예시처럼 특정 분기에서 나뉘어 진다는 특징이 있다.

PreProcessing Opening Data

lichess에서 제공한 chess opening dataset에서 우리는 png, name, epd 컬럼을 추출하여 우리가 원하는 방식대로 변경하여 csv로 저장할 것이다.

#pipline.py

from .utils import *

class ChessOpeningPipeline:
    def data_preprocessing(self, fpath):
        return import_data_csv(fpath=fpath)

    def export_to_csv(self, fname, data):
      data.to_csv(f'C:\\Users\\aaa57\\chess\\dist\\{fname}.csv')
      return

#utils.py
import pandas as pd


def import_data_csv(fpath, sep = '\t'):
  data = pd.read_csv(fpath, sep=sep)
  df = split_data(data)

  df = pd.DataFrame(df, columns=['pgn','name','epd'])

  return df


def split_data(data):
  df = []
  for i in data.index:
    val = data.loc[i, 'pgn'].split(' ')
    name = data.loc[i, 'name']
    epd = data.loc[i,'epd']
    k = _revmove_number(val)
    df.append([k,name,epd])

  df.sort(key=lambda x: len(x[0]))

  return df


def _revmove_number(value):
  a = list(filter(None ,map(_is_number, value)))

  return a

def _is_number(value):
  if value.endswith('.'):
    return 
  else:
    return value

#data_processor.py
from .pipelines import ChessOpeningPipeline

class DataProcessor:
  def data_processor():
    k = ChessOpeningPipeline()
    return k

간단한 코드 몇줄로 데이터를 정제할 수 있었는데, pipeline과 data_processor를 따로 정의한 이유는 후에 django에서 사용하기 위함.

,pgn,name,epd
0,['Nh3'],Amar Opening,rnbqkbnr/pppppppp/8/8/8/7N/PPPPPPPP/RNBQKB1R b KQkq -
1,['a3'],Anderssen's Opening,rnbqkbnr/pppppppp/8/8/8/P7/1PPPPPPP/RNBQKBNR b KQkq -
2,['f3'],Barnes Opening,rnbqkbnr/pppppppp/8/8/8/5P2/PPPPP1PP/RNBQKBNR b KQkq -
3,['h3'],Clemenz Opening,rnbqkbnr/pppppppp/8/8/8/7P/PPPPPPP1/RNBQKBNR b KQkq -
4,['g4'],Grob Opening,rnbqkbnr/pppppppp/8/8/6P1/8/PPPPPP1P/RNBQKBNR b KQkq -
5,['g3'],Hungarian Opening,rnbqkbnr/pppppppp/8/8/8/6P1/PPPPPP1P/RNBQKBNR b KQkq -
6,['h4'],Kádas Opening,rnbqkbnr/pppppppp/8/8/7P/8/PPPPPPP1/RNBQKBNR b KQkq -
7,['d3'],Mieses Opening,rnbqkbnr/pppppppp/8/8/8/3P4/PPP1PPPP/RNBQKBNR b KQkq -
8,['b4'],Polish Opening,rnbqkbnr/pppppppp/8/8/1P6/8/P1PPPPPP/RNBQKBNR b KQkq -
9,['c3'],Saragossa Opening,rnbqkbnr/pppppppp/8/8/8/2P5/PP1PPPPP/RNBQKBNR b KQkq -
10,['Na3'],Sodium Attack,rnbqkbnr/pppppppp/8/8/8/N7/PPPPPPPP/R1BQKBNR b KQkq -

이제 체스 오프닝 데이터를 프로세스화 시기키위한 준비는 다 되었다.

다음 포스팅은 django orm과 database를 활용하여 이 데이터를 저장하고 쿼리를 하는 포스팅을 할 예정.

빅데이터를 전체 클러스터에 분산해 안정적으로 저장하여 애플리케이션이 그 데이터를 신속하게 액세스해 분석할 수 있게함.

• HDFS는 대용량 파일들을 다루기 위해 만들어짐.

• 대용량 파일들을 작은 조각으로 나누어 클러스터 전체에 걸쳐 분산시키는데 최적화되어있다.

  • 대용량 파일 즉 빅데이터를 '데이터 블록'으로 나눔.
  • 기본값은 128MB 혹은 그 이하
  • 대용량 파일을 블록으로 나눔으로써 더 이상 하드 드라이브의 용량에 제한되지 않게한다.

• HDFS는 이 대용량 파일을 분산해서 처리.

• 대용량 파일을 쪼개 여러 컴퓨터에 걸쳐 저장했다면 각 컴퓨터는 자기에게 저장된 데이터 블록을 동시에 처리할 수 있게된다.

• 데이터를 효율적으로 엑세스할 수 있도록 데이터 블록을 처리하는 컴퓨터가 해당 블록이 저장된 곳이랑 가까운 거리에 있도록 조정함.

• 단 하나의 블록만 저장하지 않는다.

  • 모든 블록마다 두 개 이상의 복사본을 저장.
  • 어떤 컴퓨터가 다운되더라도 HDFS가 그 블록의 백업 복사본을 갖고 있는 다른 컴퓨터에서 정보를 불러올 수 있도록 하기 위함.

• HDFS는 클러스터의 수많은 컴퓨터에 걸쳐 분산되어 있고 실패 회복력과 큰 가능성이 있지만 결국에는 파일 시스템을 갖춘 거대한 하드 드라이브.

Name Node

• 어느 블록이 어디에 있는지를 추적.
• 파일의 이름과 HDFS 내의 가상 디렉터리 구조 등의 정보가 있고 그 파일과 관련된 모든 블록과 해당 복사본들이 어떤 노드에 저장되어 있는지를 기록.
• 어떤 게 수정되고 저장되는지 등의 정보가 Edit log에 기록됨.
• 모든 '데이터 노드'(Data Node)에 무엇이 있는지는 이름 노드에 기록되고 클라이언트 애플리케이션은 '이름 노드'(Name Node)에 쿼리해 어디로 가야할지 알아낸 다음 궁극적으로 그 '데이터 노드'에 엑세스함.

Data Node

• 실제 파일의 블록을 저장하고 있는 곳.
• Name Node와 블록의 복사본을 잘 유지하기 위해 서로 소통함.

In HDFS Read File

• '이름 노드'에게 파일 A가 필요하다고 전송.
• '이름 노드'는 파일 A는 이러이러한 '데이터 노드'의 요러요러한 블록들에 저장되어 있다고 알려줌.
• 클라이언트 애플리케이션은 해당 '데이터 노드'에 방문해 그 블록들을 가져와 파일 A를 구성.

'이름 노드'에게 그 파일이 어떤 '데이터 노드'의 어느 블록에 있는지 어떤 '데이터 노드'에 가는 게 가장 효율적인지 등을 물어보면 '이름 노드'가 어디 어디로 가야 한다고 알려준다는 개념

How To Write To HDFS

• 클라이언트 애플리케이션이 HDFS에 새 파일을 만들려고 할 때 먼저 블록의 위치를 추적하는 '이름 노드'에게 요청.
• '이름 노드'는 새 파일 항목을 확인하고 그 블록의 위치를 정한 후에 클라이언트에게 어느 '데이터 노드'에 가서 저장하라고 알려줌.
• 클라이언트는 정해진 '데이터 노드'에게 가서 파일을 건네줌.
• 그 '데이터 노드'는 주변의 다른 '데이터 노드'에 복사본을 전달하고 또 전달.

'이름 노드'에서 전달받은 정보를 클라이언트가 '데이터 노드'에게 전달하고 그 '데이터 노드'는 '데이터 노드'끼리 전달. 그리고 데이터가 저장되면 잘 받았다는 신호를 클라이언트를 통해 '이름 노드'로 보냄. 그러면 '이름 노드'는 새로운 파일의 블록과 복사본의 위치를 기억함.

• '이름 노드'는 그 파일이 존재하며 여러 '데이터 노드'에 걸쳐 복제되어 저장되었음을 기록.

When Name Node is Unavailable

• 메타데이터를 계속 백업.

• 이차적 이름 노드를 운영.

  • '동적 백업'(hot backup)이 아님. 동시에 운영하지 않기 때문.
  • '주 이름 노드'의 편집 로그 복사본을 유지하는것.

• HDFS 연합(Federation).

  • HDFS는 많은 수의 작은 파일보다는 대용량 파일을 다루는데 최적화돼 있음.
  • 이름 노드는 한계점에 다다랐을 때, HDFS 연합은 HDFS 파일 구조 내에 '명칭 공간 볼륨'(namespace volume)이라고 부르는 서브디렉터리마다 분리된 이름 노드를 지정함.
  • 그러면 각 볼륨마다 데이터 파일을 읽거나 쓸 때 어떤 이름 노드와 얘기해야 하는지 알 수 있음.

• 'HDFS 고가용성(High Availability)'을 사용해 '동적 예비 이름 노드'를 운영.

  • 공유 편집 로그를 활용하여 이름 노드가 HDFS가 아닌 다른 안전한 공유 저장소에 편집 로그를 작성.
  • 주 이름 노드가 다운되면 '동적 예비 이름 노드'가 바로 업무를 이어받음.
  • Zookeeper가 어떤 이름 노드를 사용한다고 하면 다른 이름 노드의 전원을 완전히 차단해버려 어떤 클라이언트도 이 노드와 소통하지 못하게 함.

    HDFS 고가용성을 사용할 때 예로 두 개의 이름 노드가 동시에 작동할 수도 있음. 쓰기 요청을 받을 때 한 이름 노드는 그 데이터가 어디에 있는지 알고 다른 노드는 모를 수 있는 문제가 생길 수 있음.

How To Use HDFS

• 유저 인터페이스를 사용.
• 명령줄 인터페이스.
• HTTP.
  • 직접적으로 하거나 프락시 서버를 활용.
  • 프락시 서버는 HDFS 머신과 클라이언트 사이에 위치.
  • 데이터를 분석하기 위해 스크립트를 작성하거나 애플리케이션 개발을 위해 이 프락시 서버에 웹 인터페이스를 구성.

Hive는 Hadoop 클러스터를 관계형 데이터베이스처럼 사용하지만 실제로 데이터베이스를 갖는 것은 아니었다. 그러나 만약 MySQL과 같은 진짜 관계형 데이터베이스를 갖고 있고 Hadoop 클러스터에 데이터를 불러오거나 내보내고 싶을 수도 있을 때 Sqoop을 활용하여 할 수 있다.

MYSQL

• 거대한 하드 드라이브에 연결되는 하나의 서버에 설치.
• 클러스터에 분산되지 않고 모든 것이 로컬에 있으므로 온라인 처리 프로세싱(OLTP)에 적절.
• 데이터베이스에서 빠르게 결과를 구해야 한다면 MySQL 같은 도구가 적절

Sqoop

• SQL과 Hadoop을 합친 이름.
• 대용량 데이터 세트를 Hadoop 클러스터에 가져오거나 내보내고 관리한다.
  • MapReduce 작업을 통해 가능 함.
  • 데이터베이스를 Hadoop으로 내보내려고 하면 Sqoop은 내부적으로 여러 매퍼를 작동한다.(데이터를 한 곳에서 다른 곳으로 옮기는 일이니 매퍼만 사용)
  • 매퍼들은 Hadoop 클러스터의 HDFS 클러스터와 소통하며 HDFS에 거대한 테이블 생성한다.

Apache OverView (https://blogs.apache.org/sqoop/entry/apache_sqoop_overview)

• Mapper에서 HDFS로 들어가는 과정은 병렬적으로 처리됨.
• 각 Mapper는 같은 호스트에 있는 HDFS의 블록들과 개별적으로 소통하고 있으니 병목현상이 일어나지 않음.

SQL to HDFS

sqoop import --connect jdbc:mysql://<hostname>/<databasename> -- driver com.mysql.jdbc.Driver --table <tablename>

• sqoop은 명령창에서만 작동됨.
  • {sqoop import}는 데이터를 클러스터로 가져온다는 뜻.
  • 'driver'를 명시해서 문제를 사전에 방지.

SQL to Hive

• 파일을 HDFS 클러스터로 옮기는 중간 단계를 건너 뛰고 Hive 테이블을 만드는 것.

sqoop import --connect jdbc:mysql://<hostname>/<databasename> -- driver com.mysql.jdbc.Driver --table <tablename> --hive-import

• 끝에 '--hive-import'라는 매개 변수를 추가하기만 하면 된다.

Incremental import

• 데이터베이스를 불러올 때 순차 번호나 타임스탬프를 사용해 어디까지 가져왔는지 기억할 수 있음.
• '--check-column'이란 매개 변수는 타임스탬프나 순차 번호 등을 가진 열(column).
• '--last-value'는 열(column)의 값이 지정된 값보다 클 때만 데이터를 가져오겠다는 뜻
• 이걸 사용해 데이터베이스를 불러올 때 순차 번호나 타임스탬프를 사용해어디까지 가져왔는지 기억할 수 있다.

Hive to SQL

sqoop export --connect jdbc:mysql://<hostname>/<databasename> -- driver com.mysql.jdbc.Driver --table <tablename> --export-dir /<where>/<is>/<hive data> --input-fields-terminated-by '\0001'

• 'import' 대신 {sqoop export}라고 입력하고 연결하려는 데이터베이스를 지정
• '-m 1'은 하나의 매퍼만 사용한다는 뜻.
• 데이터를 저장할 MySQL의 테이블을 지정한다 이때 Sqoop이 테이블을 따로 만들지 않도록 그 테이블이 존재하는지 확인해야 한다.

Shcema on Read

• 관계형 데이터베이스에서는 Schema on Write 개념을 사용한다.

  • 데이터베이스에 데이터를 입력하기전에 스키마를 정의.
  • 데이터를 디스크에 저장할 때 이 스키마대로 시행.

• Hive는 구조화되지 않은 데이터를 가져와서 읽는 순간에 스키마를 적용한다.

  • ex) 데이터는 탭으로 구분된 텍스트 파일이지만 실제로 구조화되지는 않은 상태
  • 데이터가 뭘 나타내는지에 대한 정보가 없다. 하지만 Hive는 그 구조화되지 않은 데이터와 연관된 실제 스키마 데이터인 메타스토어를 가지고 있음.
  • 텍스트 파일을 어떻게 해석할지 알려준다.

CREATE TABLE ratings (
    userID INT,
    movieID INT,
    rating INT,
    time INT)
ROW FORMAT DELIMETED
FIELDS TERMINATED BY '\t'
STORED AS TEXTFILE

LOAD DATA LOCAL INPAHT '<PATH the data file>'
OVERWRITE INTO TABLE ratings

• 'CREATE TABLE ratings'라는 명령어를 입력하고 다양한 필드 이름과 유형을 지정.
• 데이터가 디스크에 어떻게 구조화되는지 지정.
• 행은 탭 구분 기호로 나눠져 있고 텍스트 파일로 저장.

Hive는 구조화된 관계형 데이터베이스를 만드는 게 아님. 클러스터에 있는 기존의 데이터를 읽을 때 스키마를 적용하는 것. 'LOAD DATA' 명령어를 사용하면 데이터를 분산 파일 시스템에서 Hive로 가져오고 그 데이터에 대한 소유권을 갖고 관리한다.

Managed TABLE vs External TABLE

• 누가 데이터를 소유하고 삭제하는가의 차이.

• Managed TABLE
  • Hive로 가져온 데이터는 Hive의 소유가 되는 것.
  • 'DROP TABLE' 명령어를 사용하면 그 데이터는 삭제.
• External TABLE
  • External TABLE의 의미는 Hive가 소유권을 갖지 않는다는 겁
  • Hive 외부의 다른 시스템과 공유해야 할 때.
  • 'CREATE EXTERNAL TABLE'로 사용가능.
  • 이 데이터에 Hive를 사용하지만 소유하지는 않는다고 하는 것.
  • 'DROP'하면 메타데이터만 버리고 이 데이터는 삭제되지 않음.

Partitioning

• 만약 대용량 데이터 세트 중의 일정 부분에만 쿼리한다면 최적화하는데 파티션이 중요한 역할을 함.

CREATE TABLE customers(
    name STRING,
    address STRUCT<street:STRING, city:STRING, state:STRING, zip:INT>)

PARTITIONED BY (country STRING);

• customers 데이터베이스 안에 name, address속성이 있음.
• PARTITIONED BY (country STRING);
  • country를 하나의 열로 취급.
• 전체 데이터베이스를 검색하지 않아도 되니 많은 시간을 절약할 수 있음.

STRUCT

• 'address'를 보면 'street:STRING'과'city:STRING','state:STRING', 'zip:INT'를 포함.
  • Hive 데이터베이스에 구조화 데이터를 저장할 수 있음.

Hive

Hive 기술을 통해 hadoop 클러스터를 관계형 데이터베이스처럼 사용할 수 있다.

• SQL을 사용해 Hadoop 클러스터에 걸쳐있는 데이터를 쿼리하는 간단하고 강력한 도구.(구조화된 쿼리 언어)
• HDFS 클러스터 전체에 걸쳐 저장된 데이터에 표준 SQL 쿼리를 실행.
  • SQL을 MapReduce나 TEZ 명령어로 번역해서 작동.
  • YARN 클러스터 관리자 위에서 실행.
  • Hive는 SQL 쿼리를 매퍼와 리듀서로 분해하고 클러스터 전체에 걸쳐 어떻게 실행할지를 알아낸다.

    사용자의 관점에서는 데이터 웨어하우스를 사용하는 것처럼 SQL 데이터베이스에 쿼리하는 것과 같다.

HiveQL

• MySQL 구문과 굉장히 비슷.
• SQL에서 할 수 있는 모든 것을 HiveQL에서 할 수 있다.
• HiveQL은 대화형 명령 프롬프트를 사용.
  • 'SELECT' 명령어 등을 사용해 일반 데이터베이스에 하듯 사용가능.
• 결과가 즉각적으로 나오진 않겠지만 상식적인 시간 내에 결과를 얻을 것이고 단일 데이터베이스 인스턴스에서보다 훨씬 큰 데이터를 처리할 수 있다.
• 실시간 처리 등의 작업에는 적합하지 않을 수 있지만 대용량 데이터 세트에 분석 쿼리를 하려는 것이라면 Hive가 제격.

Hive weakpoint

• 고 처리량, 저 대기 시간을 원한다면 좋은 선택이 아님.
  • SQL 명령어를 MapReduce로 번역을 해야 하기 때문에 시간이 더 걸리기 때문.
• 비정규화 데이터를 사용.
  • 관계형 데이터베이스인 것처럼 쿼리할 수 있지만 실제론 모든 것이 비정규화되어 있고 연결된 별도의 테이블이나 키 등을 갖고 있지 않는다.
  • 업데이트나, 삽입, 삭제 등을 해도 기록되지 않는다. (실제 데이터베이스가 아니기 때문. 거대한 하나의 텍스트 파일)
• 잡한 쿼리를 해야 한다면 Pig나 Spark가 좀 더 적합.
  • (다만 나중에 VIEW를 통해 연계 사용 가능.)

Hive VIEW

• 어떤 쿼리의 결과를 'VIEW'에 저장하고 그 'VIEW'를 다음 쿼리의 테이블로 사용할 수 있는 아주 유용한 기능.
• 복잡한 쿼리를 개별 쿼리로 분할해 이 'VIEW'를 점점 발전시키며 층층이 쌓아 구축할 수 있다.

Hive example

CREATE VIEW IF NOT EXISTS topMovieIDs AS 
SELECT movieID, count(movieID) as ratingCount
FROM ratings
GROUP BY movieID
ORDER BY ratingCount DESC;

SELECT n.name ratingCount
FROM topMovieIDs t JOIN movie_names n ON t.movieID * n.movie_ID;

• 기본적으로 관계형 데이터베이스 언어 (MYSQL)의 문법과 별반 차이가 없다.
• CREATE VIEW
  • 위에서 설명한 Hive의 특징으로 VIEW를 만들어 다음 쿼리로 넘길 수 있다.

Zookeeper

• Zookeeper는 클러스터의 모든 것을 조직화하는 기술.
• 어떤 노드가 살아있는지 추적할 수 있고 여러 애플리케이션이 사용하는 클러스터의 공유 상태를 안정적으로 확인할 수 있다.
• 많은 애플리케이션이 Zookeeper에 의존.
  • 그래서 어떤 노드가 다운되더라도 일관성 있고 안정적인 성능을 온 클러스터에 걸쳐 유지할 수 있게함.
• 어떤 것이 마스터 노드이며 어떤 노드가 살아있고 다운되어 있는지 추적하는데 사용할 수 있다.

Sqoop

• Sqoop은 Hadoop의 데이터베이스를 관계형 데이터베이스로 엮어낸다.
• ODBC나 JDBC로 소통 가능한 데이터는 Sqoop을 통해 HDFS의 파일로 변형할 수 있다.
• Sqoop은 레거시 데이터베이스와 Hadoop을 잇는 연결 장치.

FLUME

• FLUME을 가지고 대규모 웹로그를 안정적으로 클러스터에 불러올 수 있다.

  예로 웹 서버 여러 개를 가지고 있다고 합시다 FLUME은 실시간으로 웹 서버의 웹로그를 감시하고 클러스터에 게시해 STORM이나 Spark Streaming을 사용해 처리한다.

Kafka

• Kafka도 데이터 수집을 하지만 좀 더 포괄적으로 사용된다.
• PC 혹은 웹 서버 클러스터에서 모든 종류의 데이터를 수집해 Hadoop 클러스터로 보낸다.

Mesos

• 클러스터의 리소스를 관리하는 또 하나의 방법.
• YARN과 협업이 가능하다.

Spark

• Hadoop 생태계에서 가장 흥미로운 기술.
• Spark는 YARN이나 Mesos중 어느 쪽을 기반으로 하든 데이터에 쿼리를 실행할 수 있음.
• Pyhon, Java, Scala를 사용해 스크립트를 작성.
• 클러스터의 데이터를 신속하고 효율적으로 처리가 가능하다.
• 많은 다양성을 갖고있음.
  • 클러스터에 걸친 정보로 머신 러닝을 수행하는 SQL쿼리.
  • 실시간으로 스트리밍되는 데이터를 처리하는 등.

TEZ

• 방향성 비사이클 그래프
  • TEZ는 MapReduce의 일을 할 때 더 유리함.
  • 쿼리실행에 더 효율적인 계획을 세우기 때문.
  • Hive와 함께 사용하면 성능을 극대화 시킬 수 있음.

HBASE

• HBASE는 클러스터의 데이터를 트랜잭션 플랫폼으로 노출하는 역할을 하며 NoSQL 데이터베이스라고 불린다.
• 기둥형 데이터 스토어
  • 단위 시간당 실행되는 트랜잭션의 수가 큰 아주 빠른 데이터베이스
  • 데이터를 웹 애플리케이션이나 웹사이트에 노출시켜 OLTP 트랜잭션을 하는데 적합하다.
• HBASE는 클러스터에 저장된 데이터를 노출시킨다.
• 데이터는 Spark나 MapReduce 등에 의해 전환되었을 수도 있고 후에 그 결과를 다른 시스템에 노출시킬 빠른 방법을 제공한다.

Apache STORM

• STORM은 스트리밍 데이터를 처리하는 방식.
• 센서나 웹로그로부터 데이터를 스트리밍 한다면 STORM이나 'Spark Streaming'을 통해 실시간으로 처리할 수 있다.
• 일괄로 처리할 필요가 없음.
• 데이터가 실시간으로 입력됨에 따라 실시간으로 '기계 학습'을 업데이트하거나 데이터를 데이터베이스에 저장할 수 있다.

OOZIE

• OOZIE는 일정에 따라 작업을 순차적으로 진행할 수 있도록 스케줄링.

  예를 들어 데이터를 Hive에 불러와서 Pig를 통해 통합하고 Spark를 통해 쿼리한 후에 결과를 HBASE로 변환시킨다고 하면 OOZIE가 이 모든 것을 관리해 안정적이고 일관성 있게 실행할 수 있다.

HDFS

• Hadoop 분산 파일 시스템
• 빅데이터를 클러스터의 컴퓨터들에 분산 저장하는 시스템.
• 데이터의 여분 복사본까지 만들어 데이터의 손실을 자동을 회복함.
• Hadoop의 데이터 저장 부분.

YARN

• 'Yet Another Resource Negotiator'의 약어이며 '또 다른 리소스 교섭자'라는 뜻.
• YARN은 Hadoop의 데이터 처리 부분.
• 컴퓨터 클러스터의 리소스를 관리하는 시스템.
  • 누가 작업을 언제 실행하고 어떤 노드가 추가 작업을 할 수 있고 누구는 할 수 없고 등을 결정.

MapReduce

• 데이터를 클러스터 전체에 걸쳐 처리하도록 하는 프로그래밍 모델.
• Mapper와 Reducer로 구성되어 있어 프로그램을 사용할 때 사용하는 두 개의 구분된 함수
  • Mapper는 클러스터에 분산돼있는 데이터를 효율적으로 동시에 변경할 수 있음.
  • Reducer는 그 데이터를 집계함.

Pig

• Pig는 고수준의 API'로써 많은 경우 SQL과 비슷한 간단한 스크립트를 작성해 쿼리를 연결하고 복잡한 답을 구할 수 있다.
• 프로그래밍 언어를 사용하지 않아도 사용 가능.
• Pig는 작성된 스크립트를 MapReduce가 읽을 수 있도록 번역하고 MapReduce는 다시 YARN과 HDFS에게 데이터를 처리하고 원하는 답을 가져오게한다.
• Pig는 MapReduce위에 있는 고수준 스크립팅 언어.

Hive

• MapReduce위에 구축돼 Pig처럼 작동하지만 더 SQL 데이터베이스 역할을 수행함.
• 셸 클라이언트나 ODBC(Open Database Connectivity) 등을 통해 데이터베이스에 접속할 수 있으며 Hadoop 클러스터에 저장돼있는 데이터가 내부적으로는 관계형 데이터베이스가 아님에도 불구하고 SQL로 쿼리한다

Hadoop 플랫폼의 주요 벤더인 Hortonworks는 하둡을 이렇게 정의.

범용 하드웨어로 구축된 컴퓨터 클러스터의 아주 방대한 데이터 세트틀 분산해 저장하고 처리하는 오픈 소스 소프트웨어 플랫폼

Hadoop

• 오픈 소스
• 한대의 PC가 아닌 컴퓨터 클러스터에서 작동하는 소프트웨어 묶음
  • Hadoop은 어느 데이터 센터에서 작동중인 컴퓨터 클러스터 전체에서 실행되도록 설계됨.
  • 즉 Hadoop은 다수의 PC를 활용해 빅데이터를 다룬다.

• 분산저장
  • 분산 저장은 Hadoop이 제공하는 주 기능.
  • 한 개의 하드 드라이브에 국한되지 않는다.
  • 빅데이터(1TB)를 처리할 때 클러스터에 컴퓨터를 더하기만 하면 그 컴퓨터는 하드 드라이브가 데이터 저장소의 일부가 됨.
  • Hadoop은 클러스터의 모든 하드 드라이브에 걸쳐 분산돼 있는 모든 데이터를 단일 파일 시스템으로 보여줌.

• 데이터의 여분 제공
  • Hadoop은 클러스터의 컴퓨터 중 하나가 사용 불가 상태에 놓여있어도 이런 상황을 데이터 백업 복사본을 클러스터의 다른 컴퓨터에도 보관하기 때문에 자동으로 소실된 데이터를 복구할 수 있음.

• 분산 처리
  • Hadoop은 데이터를 다른 양식으로 전환하거나 다른 시스템으로 전송할 때 혹은 집계해야 할 때 모든 작업을 병렬로 처리한다.
  • 클러스터 내 모든 컴퓨터 CPU에게 이 작업을 분배하여 동시에 처리.
  • 많은 데이터를 신속히 처리할 수 있음.
  • 처리할 수 없다면 컴퓨터를 추가하면 가능.