[Hadoop] What is Pig?

1. What is Pig ?

Apache Pig는 대규모 비정형 및 반정형 데이터를 분석하기 위한 프레임워크이다.

Pig는 구조화되지 않은 데이터들을 처리하는데 적합한 언어로, 테이블 구조가 아니더라도 데이터 처리 가능 하며 PigLatin이라는 언어를 사용하여 SQL과 유사한 방식으로 데이터를 처리할 수 있습니다.

Hadoop 클러스터에서 실행되며, 데이터 처리를 위해 MapReduce를 사용합니다.

Pig는 대규모 데이터 처리를 단순화하고 생산성을 높이는 데 효과적입니다.

 

2. Pig 특징

1) Pig Latin 언어 with 실행 계획

• Pig는 Pig Latin이라는 고급 스크립팅 언어를 사용하여 데이터 변환, 필터링, 집계 등의 작업을 쉽게 수행할 수 있습니다. SQL과 유사한 문법을 제공해 데이터 작업을 직관적으로 처리할 수 있습니다.

    => Pig Latin으로 작성한 데이터 처리 프로그램은 논리적인 실행 계획으로 변환되고, 이것은 최종적으로 MapReduce 실행 계획으로 변환

Pig의 동작 과정 예시

2) 추상화된 데이터 처리

• Pig는 데이터를 로드하고, 변형하고, 저장하는 등의 작업을 고수준에서 처리할 수 있도록 해줍니다. 이를 통해 사용자는 더 적은 노력으로 복잡한 데이터 분석 작업을 수행할 수 있습니다. 즉, 스크립트가 내부적으로 MapReduce를 수행하기에 사용자는 더 직관적으로 데이터를 다루기 쉬움

3) Pig는 Hive에 비해 JOIN 기능에 있어서 성능적으로 좋다. 그 이유는 여러 개의 테이블을 조인하여 MapReduce하게 된다면 셔플링 작업에 성능적인 제한사항이 있다. 실행계획과 함께 작업을 실행하면 해결 할 수 있지만 Hive에서는 약하다. 그에 비해 Pig에서는 실행 계획이 있기에 여러 대안 중 하나를 선택하여 최적화를 진행해준다. 즉, 옵티마이저를 실행 해주기에 장점도 있지만 잘못 실행하면 반복을 많이하고 CPU를 많이 사용하게 된다. 

 

4) Pig는 소량의 데이터에는 비효율적입니다.

 

• MapReduce의 초기화 및 오버헤드 => 즉, 작은 데이터에 대해서도 Pig는 Hadoop 클러스터에서 작업을 분배하고, 데이터 분할, 맵 작업, 리듀스 작업 등의 단계를 거치기 때문에, 처리 시간이 상대적으로 오래 걸립니다.
• 배치 처리(batch processing) 방식 =>. 배치 처리는 데이터를 모아서 한 번에 처리하는 방식으로, 대량 데이터를 효율적으로 처리하는 데 적합합니다. 하지만 소량의 데이터의 경우 이러한 배치 처리 방식은 지연 시간(latency)을 증가시킬 수 있습니다.

5) Pig 사용하는 기업 : Yahoo, Zoom

 

3. Pig 사용 예시

문제 : 연도별 평균 발생 횟수가 가장 많은 20개의 바이그램을 해당 평균값과 구하기.
데이터 : Google book bigram (1-gram)

# Load a-1gram, b-1gram 파일 LOAD
data = LOAD 'hdfs://cluster-d37a-m/test/googlebooks-eng-all-1gram-20120701-b' USING PigStorage('\t') AS (bigram, year, match_count, volume_count); 
data2 = LOAD 'hdfs://cluster-d37a-m/test/googlebooks-eng-all-1gram-20120701-a' USING PigStorage('\t') AS (bigram, year, match_count, volume_count); 

# 두 파일 JOIN 기능 수행
join_data = UNION data, data2; 

# bigram 에 따라 GROUP 기능 수행
grouped_data = GROUP join_data BY bigram; 

# SUM 및 COUNT 기능 수행
bigram_status = FOREACH grouped_data GENERATE group AS bigram,  
SUM(join_data.match_count) AS total_match_count,  
COUNT(join_data.year) AS num_years; 

# avg_match_per_year 값 계산
res_data = FOREACH bigram_status GENERATE bigram,  
(double)total_match_count / num_years AS avg_match_per_year; 

# ORDER BY 기능 수행
sorted_data = ORDER res_data BY avg_match_per_year DESC; 
top_20_bigrams = LIMIT sorted_data 20; 

# STORE 기능 수행
STORE top_20_bigrams INTO 'hdfs://cluster-d37a-m/test/output2' USING PigStorage('\t');

 

각 단계의 수행 방식과 Hadoop 분산 시스템에서의 동작 과정

1. LOAD => HDFS에서 데이터를 로드하며, Pig는 이를 처리하기 위해 내부적으로 MapReduce 작업을 준비
2. JOIN => Pig는 결합된 데이터를 기반으로 다음 단계의 작업을 준비하며, 이 단계에서 두 파일의 데이터를 하나로 병합하는 MapReduce 작업이 수행됩니다.
3. GROUP => bigram 값을 기준으로 데이터가 분산되고, 같은 bigram 값을 가진 데이터가 하나의 그룹으로 모입니다. 이 과정에서 Pig는 데이터를 정렬하고 그룹화하는 MapReduce 작업을 실행합니다.
4. SUM / COUNT => FOREACH를 사용해 각 bigram 그룹에서 계산을 수행합니다. 각 그룹의 match_count 값을 모두 더해 total_match_count를 계산하고, 해당 bigram이 등장한 연도의 개수(year)를 세어 num_years를 계산하기에 MapReduce 작업을 사용하여 병렬 처리를 수행
5. 새로운 필드 값 생성 => 각 bigram 그룹에 대해, 등장 횟수(total_match_count)를 연도 개수(num_years)로 나누어 avg_match_per_year(연도별 평균 등장 횟수)를 계산, 이 또한 MapReduce 작업을 통해 처리됩니다.
6. ORDER BY / LIMIT => 계산된 결과를 avg_match_per_year 기준으로 내림차순으로 정렬한 후, 상위 20개의 bigram만 추출합니다.
7. STORE => STORE는 최종적으로 데이터를 지정된 경로에 저장하며, HDFS에 데이터를 기록하는 작업을 수행합니다. 이때도 분산 파일 시스템에 데이터를 기록하기 위해 MapReduce 작업이 진행됩니다.

 

Pig 실행 결과

 

 

연간 평균 발생 횟수가 가장 많은 20개의 bigram 출력