내맘대로 뉴스 요약 큐레이션 (2) - AWS Lambda 로 크롤링 자동화하기
데이터분석 하기 전에 최근일자까지 데이터를 다시 크롤링해와야하는 귀찮음이 있어서 이번주는 로컬에서 손으로 돌리던 크롤링을 자동화하기로 했다.
0. 어떻게 만들어 볼까?
개인 aws 계정을 사용할 예정이기 때문에, 다른건 다 필요없고 가장 저렴하게 자동화된 데이터 수집 파이프라인을 만들고 싶었다.
찾아보니 AWS Cloud watch Event Trigger -> Lambda function -> S3에 저장 으로 하는것이 가장 경제적이라 카더라는 얘기가 많아서 이렇게 만들어보기로 했다.
1. AWS Lambda Function 만들기
Lambda 함수를 만들러 가보자.
Lambda 서비스에서 함수 생성 기능을 누르면 아래와 같이 만들려는 함수 이름, 런타임, 권한에 대해서 지정할 수 있다.
런타임은 나중에 필요한 selenium + chromium 때문에 python 3.6 버전으로 설정했다.
함수생성을 하면 아래와 같은 화면을 볼 수 있다.
트리거 추가에서 lambda function 을 실행할 때 필요한 trigger 를 추가할 수 있고, 함수 코드 영역에 실행할 코드를 직접 작성하거나, 압축된 실행 파일을 업로드할 수 있다.
1) Layers 에 Selenium + Chromium 패키지 찾아서 올리기
예전에는 lambda 코드에서 python 기본 라이브러리 외 다른 라이브러리를 사용하는 경우, 스크립트 코드와 스크립트에 필요한 파이썬 패키지파일들을 모두 한데모아…. zip으로 압축해서 올려주어야 했다.
(lambda_function.zip…)
그런데 최근에는 Layers 라는 기능이 추가되어서 필요한 패키지 파일을 따로 Layer에 넣어둘 수 있고, 스크립트 코드에서 Layer 에 있는 패키지를 import 할 수 있도록 분리되었다. (오예!)
사실 예전에 lambda 함수를 써보려고 할 때 테스트 환경이 불편해서 잘 쓰지 못했었다. 특히 외부 라이브러리를 쓰는 경우나, lambda 실행 환경에서 크롤링에 필요한 headless chrome driver를 테스트 해보는 경우 더더욱 시간이 많이 들고 어려웠다. 그런데 Layers 기능이 생기면서 Layer에 필요한 패키지를 올려두고 스크립트 코드만 작성할 수 있게 되어서 빠른 테스트도 가능해졌다.
찾아보니 로컬에서 테스트할 때는 AWS lambda 의 실행환경에 맞춘 lambci 에서 제공하는 docker 이미지 파일로 테스트 해볼 순 있다고 한다. 하지만 그냥 바로 editor 에서 테스트하는게 제일 편하다.
그러면 나에게 필요한 건 크롤링 코드를 돌리기 위한 selenium + chrome driver Layer 가 필요하다.
lambda 를 이용해서 selenium 으로 크롤링하는 건 이전에 매우 많은 훌륭하신 분들이 많은 삽질을 해 두었을 것이므로, 열심히 구글링해보니 바로 구할 수 있었다.
검색해서 나온 여러 결과 중 https://github.com/inahjeon/AWS-LAMBDA-LAYER-Selenium 에 올려진 파일을 사용했다.
SELENIUM_LAYER.zip 을 받아서 Layer 에 등록하자.
Layers -> create layer 에서 zip 파일만 업로드 해주면 등록되고, lambda function 에서 add a layer 로 등록한 layer 를 선택해서 추가해주면 된다.
크롤링 코드는 실행 시간이 좀 필요하기 때문에 timeout 설정을 1분 이상 지정해주자.
import 가 잘된다! 오늘 하는 것 중 가장 난이도 어려운 부분이 쉽게 해결되었다! (덩실덩실)
2) Cloud Watch Event Trigger 설정하기
이벤트 트리거는 add trigger 로 들어가서 CloudWatch Events/EventBridge 를 선택하고 아래와 같이 만들 수 있다.
매일 새벽 1시에 실행되도록 했다.
3) s3에 결과 저장하기
이제 s3에 크롤링한 결과를 저장해보자.
일단 lambda function 에 연결된 권한에 s3 접근 권한도 추가해주어야 한다.
IAM -> 역할 에서 연결된 역할을 선택하고 정책연결을 눌러 AmazonS3FullAccess 권한을 추가 해준다. (세밀한 권한 설정은 잘 모르므로 패스)
스크립트에서 웹페이지를 s3에 저장하는 코드를 테스트해보자.
def lambda_handler(event, context):
driver = get_driver()
driver.get('https://www.google.com/')
page_data = driver.page_source
driver.close()
encoded_string = page_data.encode("utf-8")
file_name = f"test.txt"
s3_path = f"test/" + file_name
s3 = boto3.resource("s3")
s3.Bucket(BUCKET_NAME).put_object(Key=s3_path, Body=encoded_string)
return page_data
크롤링한 데이터가 잘 저장된다.
이제 필요한 기능들은 다 테스트했고, 이전에 작성해 둔 크롤링 코드만 적절히 변경해서 넣어주면 된다.
2. 뉴스 크롤링 하기
이전 글에서 작성해 두었던 코드를 조금 변경하여 매일매일 그날의 인기뉴스 리스트를 가져와서 s3에 저장하는 코드로 변경했다.
import os
from datetime import datetime, timezone, timedelta
import boto3
from selenium.webdriver.chrome.options import Options
from selenium import webdriver
BUCKET_NAME = os.getenv('BUCKET_NAME', '')
def get_driver():
chrome_options = Options()
chrome_options.add_argument('--headless')
chrome_options.add_argument('--no-sandbox')
chrome_options.add_argument('--disable-gpu')
chrome_options.add_argument('--window-size=1280x1696')
chrome_options.add_argument('--user-data-dir=/tmp/user-data')
chrome_options.add_argument('--hide-scrollbars')
chrome_options.add_argument('--enable-logging')
chrome_options.add_argument('--log-level=0')
chrome_options.add_argument('--v=99')
chrome_options.add_argument('--single-process')
chrome_options.add_argument('--data-path=/tmp/data-path')
chrome_options.add_argument('--ignore-certificate-errors')
chrome_options.add_argument('--homedir=/tmp')
chrome_options.add_argument('--disk-cache-dir=/tmp/cache-dir')
chrome_options.add_argument('user-agent=Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/61.0.3163.100 Safari/537.36')
chrome_options.binary_location = "/opt/python/bin/headless-chromium"
driver = webdriver.Chrome('/opt/python/bin/chromedriver', chrome_options=chrome_options)
return driver
_SECTIONS = {
'100': '정치',
'101': '경제',
'102': '사회',
'103': '생활/문화',
'104': '세계',
'105': 'IT/과학'
}
def write_to_s3(date_str, news_type, data: str):
encoded_string = data.encode('utf-8')
file_name = f'{date_str}.txt'
s3_path = f'{news_type}/' + file_name
s3 = boto3.resource('s3')
s3.Bucket(BUCKET_NAME).put_object(Key=s3_path, Body=encoded_string)
def get_popular_day_news_contents(driver, date_str, section_id):
url = f'https://news.naver.com/main/ranking/popularDay.nhn' \
f'?rankingType=popular_day' \
f'§ionId={section_id}&date={date_str}'
driver.get(url)
driver.implicitly_wait(1)
ranking = driver.find_elements_by_class_name('ranking')
if ranking:
headlines = ranking[0].find_elements_by_class_name('ranking_headline')
ledes = ranking[0].find_elements_by_class_name('ranking_lede')
offices = ranking[0].find_elements_by_class_name('ranking_office')
views = ranking[0].find_elements_by_class_name('ranking_view')
return zip(headlines, ledes, offices, views)
else:
return None
def lambda_handler(event, context):
tz = timezone(timedelta(hours=9))
kst_dt = datetime.now().astimezone(tz)
date_str = kst_dt.strftime('%Y%m%d')
driver = get_driver()
columns = ['date', 'section', 'office', 'view', 'headline', 'lede', 'link']
data = []
data.append('\t'.join(columns) + '\n')
for section_id, section_name in _SECTIONS.items():
contents = get_popular_day_news_contents(driver, date_str, section_id)
if contents:
for headline, lede, office, view in contents:
headline_a_tag = headline.find_element_by_tag_name('a')
contents = '\t'.join([
date_str,
section_name,
office.text,
view.text,
headline_a_tag.get_attribute('title'),
lede.text,
headline_a_tag.get_attribute('href')
])
data.append(contents + '\n')
write_to_s3(date_str, 'popular_day', ''.join(data))
driver.close()
3. 결과물 확인
Test로 Cloud Watch Event trigger 에서 임시로 1분 뒤인 오후 9시 40분으로 스케쥴링해서 기다려본다.
오후 9시 42분에 job이 종료되었고 s3에 데이터를 잘 저장한 걸 확인할 수 있었다.
걸린시간은 89400ms, 메모리는 340MB 정도 사용한다고 한다.
Lambda 요금을 보면 메모리 512MB에 100ms 당 0.0000008333 USD 라고 하니, 1년 365일 돌리면 $28 달러 정도다.
결과물이 매우 만족스럽다. 자동화 칭찬해. 👏👏👏