[Spark] Spark Streaming - Twitter 해시태그 분석 (2)

지난 1편에서는 실습에 쓰일 코드를 살펴보았다. 
이어서 직접 실행을 해보자. 

1편 : https://hyem207.tistory.com/57

[Spark] Spark Streaming - Twitter 해시태그 분석 (1)

---

📜 Spark Streaming으로 Twitter API의 해시태그 분석하기 : Twitter에서 kpop으로 필터링 한 데이터를 실시간으로 가져와 해시태그(#) 분석해보기

 

실습 결과 미리 보기

터미널2 실행 결과 (2)

A. 준비단계

docker hub에서 실습 도커 이미지를 다운로드 한다.

 

A-1. 도커 이미지 설명

Spark와 Hadoop 설치 및 환경변수 설정 완료

/opt 아래에 설치된 hadoop과 spark

Spark streaming을 위한 파이썬 파일들

: read_twitter.py 와 spark_twitter.py

/home/boaz/spark-streaming 디렉토리에 위치한 파일 2개

 

 

A-2. 실습 파일 설명

(1) 구성

1. read_twitter.py : twitter api에 연결하여 특정 카테고리 정보가 달린 트윗들을 socket으로 보냄 (like 데이터 보내는 서버)
2. spark_twitter.py : read_twitter와 spark streaming의 socket과 연결됨. socket으로 받은 트윗 데이터들 중 sql 문을 이용해 해시태그만 골라와 테이블에 저장함 (2초 간격) (like 데이터 받아 분석하는 클라이언트)

자세한 코드 설명은 1편에 있다.(링크)

 

B. 실습

1. 이미지 다운 및 컨테이너 실행

1. 도커 허브에서 실습 도커 이미지를 다운로드 한다.

docker pull hmk9667/structured-spark-streaming:version1

 

 2. 다운받은 이미지 확인

docker images

   

   3. 이미지 실행하여 컨테이너 만들기 (컨테이너 명 : spark-streaming)

docker run -itd --name spark-streaming hmk9667/structured-spark-streaming:version1

 

   4. 컨테이너 목록 확인하기

docker container ls

 

   5. 컨테이너 실행하기

docker exec -it spark-streaming bash

 

   ❗ 만약 ls 명령 실행 시 bash: ls: command not found 오류가 뜬다면 다음 명령어를 터미널에 입력하기❗

export PATH=/usr/bin:/bin:/usr/sbin:/sbin:/usr/local/bin:/usr/X11/b‌in

 

2. docker exec

뒤에서 동시에 파이썬 스크립트를 2개를 실행하기 위해서 터미널 및 powershell 2개를 켜서 모두 접속한다.

docker exec -it spark-streaming bash

! 만약 컨테이너가 종료된 상태라면 아래 명령어로

docker start spark-streaming

 

3. cd /home/boaz/spark-streaming

접속한 터미널 둘 다 이동한다.

cd /home/boaz/spark-streaming

 

4. 트위터 api 키 등록

twitter api 사용을 위해 터미널에서 트위터 bearer 토큰 값을 등록해준다.

토큰 값은 트위터 개발자 계정 생성 후 만든 프로젝트에서 BEARER_TOKEN값을 가져오면 된다.

 

read_twitter.py 에서 환경변수 값을 가져와 api에 연결하기 때문에 꼭 필요한 작업이다.

export BEARER_TOKEN=(트위터 bearer토큰 값)

 

5. python3 read_twitter.py

❗ 터미널1에서 read_twitter.py 부터 실행 후 터미널 2에서 spark_twitter를 실행한다 ❗

터미널1 에서 read_twitter를 실행한다.

python3 read_twitter.py

 

6. python3 spark_twitter.py

터미널 2 에서 spark_twitter를 실행한다.

❗ 터미널1에서 read_twitter.py 부터 실행 후 터미널 2에서 spark_twitter를 실행한다 ❗

python3 spark_twitter.py

 

7. 결과 확인

• 터미널 1 에서 정한 규칙으로 twitter 에서 데이터를 불러와 client_socket에 send하는 것을 확인할 수 있음
• 터미널 2 에서 실시간으로 spark streaming으로 받아온 데이터를 2초 단위로 집계해주는 것을 확인할 수 있다.

[ 터미널 1 : python3 spark_twitter.py ]

터미널1 실행 결과

ctrl+c로 강제 중지하지 않는 이상 계속해서 twitter 데이터를 불러와 client_socket에 보내는 데이터가 콘솔에 보여진다.

 

 

[ 터미널 2 : python3 spark_twitter.py ]

터미널2 실행 결과 (1) (2)

 

소켓 통신이 계속 되는 동안 스트리밍으로 받아온 데이터를 2초 간격으로 집계하여 해시태그별로 count를 하는 것을 콘솔로 확인할 수 있다.

 

📢 실습을 마친 후에는 꼭 소켓 통신을 종료하자. Twitter API사용 시, tweet을 pull 해올 수 있는 한계가 있어서 과금 방지를 위해 꼭 종료하기

 

 

 

참고

(Twitter API 연결 파트) Twitter API v2 공식 github

Twitter-API-v2-sample-code/filtered_stream.py at main · twitterdev/Twitter-API-v2-sample-code

(Spark 스트리밍 파트) structured-spark streaming

Easy to Play with Twitter Data Using Spark Structured Streaming

(소켓 연결 파트) dstream-spark-streaming

python-spark-streaming/TweetRead.ipynb at master · jleetutorial/python-spark-streaming