[GCP] Cloud Function & Cloud Scheduler를 사용한 주가 데이터 수집 1 (Cloud Function 편)

Step 1.

그리고 나서 코드 화면이 나오면 런타임은 파이썬 3.10으로 맞춰주고, 소스코드는 인라인 편집기로 맞춰주자. 이제 Storage와 Bigquery에 접속 하기 위한 Json파일을 만들어 주자. 그리고 나서 IAM에 만들어 놓은 JSON 파일에 대한 정보를 그대로 넣어 주면 된다. 이 json 파일에 대한 정보 역시 다음의 링크를 참고 하면 된다.

• BigQuery, Storage - Python 연동

Step 2.

이번에는 requirements.txt에 우리가 필요한 패키지를 넣어 주자. 이 requeirements.txt 파일은 Cloud Function에서 함수가 실행될때 이미지 형식으로 만들어지는데, 이때 한꺼번에 설치가 진행 된다.

functions-framework==3.*
pandas
finance-datareader
psycopg2
sqlalchemy
pykrx

bs4
pandas-gbq
google-cloud-bigquery
google-cloud-storage

Step 3.

이제 마지막으로 코드 부분을 작성해주자. 코드 부분에는 다음과 같은 코드를 넣어 주면 된다. 이 코드는 지난번 Airflow 블로그에서 주가 데이터 수집 파이프라인1 (국내주식편)의 코드를 가져온것.

• 주가 데이터 수집 파이프라인 1 (국내주식편)

주의사항

여기서 주의사항이 하나 있다. Cloud Function을 알고 나서 이리저리 블로깅을 하면서 자신감이 차 있는 상태에서 Google Storage에 업로드 할때 원인 모를 에러가 계속 발생하게 된다. 왜 그런지 에러를 검색해봐도 알수 없는 상태가 몇개월 가량 지속 되었었는데, 여기서 핵심은 /tmp 라는 임시파일 경로이다. Google Storage에 업로드 하기 위해서는 파이썬 객체가 아니라 csv라는 파일이 필요 한데 여기에서 /tmp 에 파일을 저장해주지 않으면 에러가 발생한다. 물론 파이썬에서 tempfile라이브러리를 사용 해도 된다. 그러니 Cloud Function에서 파일을 저장할 경우가 생긴다면 무조건 /tmp 파일에 임시로 저장해주어야 에러가 발생하지 않는다.

#!/usr/bin/env python
# coding: utf-8
import functions_framework

import pandas as pd
import pandas_gbq
from pykrx import stock
from pykrx import bond
from datetime import timedelta

from time import sleep

import psycopg2 as pg2
from sqlalchemy import create_engine

from datetime import datetime
from datetime import timedelta

import os
import time

import glob
from google.cloud import bigquery
from google.oauth2 import service_account
from google.cloud import storage



# 서비스 계정 키 JSON 파일 경로
key_path = glob.glob("./*.json")[0]

# Credentials 객체 생성
credentials = service_account.Credentials.from_service_account_file(key_path)

# 빅쿼리 정보
project_id = '프로젝트 이름'
dataset_id = '빅쿼리 데이터셋 이름'

# GCP 클라이언트 객체 생성
storage_client = storage.Client(credentials = credentials,
                         project = credentials.project_id)
bucket_name = '구글 스토리지 버킷 이름'    # 서비스 계정 생성한 bucket 이름 입력


def upload_df(data, file_name, project_id, dataset_id, time_line, today_date1):
    if not os.path.exists(f'/tmp/{file_name}'):
        os.makedirs(f'/tmp/{file_name}')

    try:
        if not os.path.exists(f'/tmp/{file_name}/{file_name}_{today_date1}.csv'):
            data.to_csv(f'/tmp/{file_name}/{file_name}_{today_date1}.csv', index=False, mode='w')
        else:
            data.to_csv(f'/tmp/{file_name}/{file_name}_{today_date1}.csv', index=False, mode='a', header=False)
        print(f'{file_name}_로컬CSV저장_success_{time_line}')
    except:
        print(f'{file_name}_로컬CSV저장_fail_{time_line}')


    # Google Storage 적재
    source_file_name = f'/tmp/{file_name}/{file_name}_{today_date1}.csv'    # GCP에 업로드할 파일 절대경로
    destination_blob_name = f'data_crawler/{file_name}/{file_name}_{today_date1}.csv'    # 업로드할 파일을 GCP에 저장할 때의 이름
    bucket = storage_client.bucket(bucket_name)
    blob = bucket.blob(destination_blob_name)
    blob.upload_from_filename(source_file_name)

    try:
        # 빅쿼리 데이터 적재
        data.to_gbq(destination_table=f'{project_id}.{dataset_id}.{file_name}',
          project_id=project_id,
          if_exists='append',
          credentials=credentials)
        print(f'{file_name}_빅쿼리저장_success_{time_line}')
    except:
        print(f'{file_name}_빅쿼리저장_fail_{time_line}')


### 날짜 설정
now = datetime.now()

today_date1 = now.strftime('%Y%m%d')
today_date2 = now.strftime('%Y-%m-%d')
start_date2 = today_date2
today_date_time_csv = now.strftime("%Y%m%d_%H%M")


file_name = 'usd_krw'
ticker_nm = 'usd_krw'
try:
    now1 = datetime.now()

    time_line = now1.strftime("%Y%m%d_%H:%M:%S")
    df_raw = fdr.DataReader('USD/KRW', start_date2,today_date2)
    df_raw['ticker'] = ticker_nm
    df_raw = df_raw.reset_index()
    df_raw.columns = ['date', 'open','high','low','close','adj_close','volume','ticker']

    print(f'{ticker_nm} success_{time_line}')
except:
    print(f'{ticker_nm} fail_{time_line}')

df_raw['date'] = pd.to_datetime(df_raw['date'])
now1 = datetime.now()

time_line = now1.strftime("%Y%m%d_%H:%M:%S")
upload_df(df_raw, file_name, project_id, dataset_id, time_line, today_date1)



@functions_framework.http
def hello_http(request):
    """HTTP Cloud Function.
    Args:
        request (flask.Request): The request object.
        <https://flask.palletsprojects.com/en/1.1.x/api/#incoming-request-data>
    Returns:
        The response text, or any set of values that can be turned into a
        Response object using `make_response`
        <https://flask.palletsprojects.com/en/1.1.x/api/#flask.make_response>.
    """
    request_json = request.get_json(silent=True)
    request_args = request.args

    if request_json and 'name' in request_json:
        name = request_json['name']
    elif request_args and 'name' in request_args:
        name = request_args['name']
    else:
        name = 'World'
    return 'Hello {}!'.format(name)

여기까지 마무리 했으면 마지막으로 배포 버튼을 클릭하면 된다. 그리고 나서 시간이 지나고 대시보드를 보면 다음과 같이 성공 화면을 확인할 수 있다.

Storage, Bigquery 확인

이제 Storage, Bigquery에서 확인하면 다음과 같이 데이터가 잘 들어온 것을 확인할 수 있다.