T아카데미 - 데이터 엔지니어링 기초 (4)Google Cloud Functions을 이용해 BigQuery에 데이터 옮기기

강의영상

 

강의 요약

• BigQuery
  • 완전관리형 데이터 웨어하우스
  • 매우 큰 데이터도 저렴하게 저장, 빠른 속도로 처리 가능
• Google Cloud Functions
  • 구글 클라우드의 서버리스 제품
  • 트리거 조건과 코드를 설정해놓으면 원하는 동작을 자동으로 수행
• 실습
  • 프로젝트 이름 클릭 → 데이터세트 만들기
  • 테이블, 스키마 생성
    • 스키마 순서는 나중에 변경할 수 없으므로 주의
  • Pub/Sub이랑 연동: Cloud 함수 트리거
  • Cloud 함수 설정
    • API 사용 설정해야 함
    • 런타임: python 3.7
    • Source code: Inline Editor
    • main.py: 트리거되었을 때 실행할 코드
    • requirements.txt: main.py 실행하는 데 필요한 패키지 정보
  • BigQuery에 저장된 데이터 확인: 쿼리 편집기
    • SELECT * FROM 테이블 LIMIT 1000

실습 코드

main.py

import base64
import json
from google.cloud import bigquery
def tweets_to_bq(tweet):
    client = bigquery.Client()
    dataset_ref = client.dataset('tweet_data')
    table_ref = dataset_ref.table('tweets')
    table = client.get_table(table_ref)
    tweet_dict = json.loads(tweet)
    rows_to_insert = [
    (tweet_dict['id'], tweet_dict['created_at'], tweet_dict['text'])
    ]
    error = client.insert_rows(table, rows_to_insert)
    print(error)

def hello_pubsub(event, context):
    """Triggered from a message on a Cloud Pub/Sub topic.
    Args:
         event (dict): Event payload.
         context (google.cloud.functions.Context): Metadata for the event.
    """
    pubsub_message = base64.b64decode(event['data']).decode('utf-8')
    print(pubsub_message)
    tweets_to_bq(pubsub_message)

requirements.txt

# Function dependencies, for example:
# package>=version
google-cloud-bigquery