brunch

라이킷 7 댓글

You can make anything
by writing

C.S.Lewis

계정을 잊어버리셨나요?

by 이민우 Jun 17. 2020

Crawler with Github Action

새 시즌을 준비하는 Professional E-Sports Player

Github Action

사내에서 진행하는 교육인 카일 스쿨 수강 중에 Github Action에 대해 알게 되었습니다.

본래 github action 사용법은 깃헙에서 일어나는 다양한 자동화를 위한 도구이고 대표적인 활용 사례로 CI/CD 등의 동작을 기대하고 제작된 것이었겠지만,

제공되는 다양한 도구와 기능, 사용법과 활용 사례들을 보고 나니, 생각보다 많은 활용 방법이 있으리라는 생각이 들었습니다. 실제로도 상당한 수의 예시를 찾아볼 수 있었습니다.

개인적으로 이러한 자동화 도구나 봇을 운영하기 위해서 전용의 인스턴스를 운영하는 것은 비용 부담이 되기 때문에 엄두조차 내지 못 했던 경우가 많았는데 공짜...인 Github Action 덕분에 그간 하지 못했거나 해보고 싶었던 것들을 생각하게 되는 계기가 되었습니다.

또한, 평소 회사에서 사용하던 Airflow와는 달리 Github Action에서의 Workflow는 비교적 단순하고 사용이 더 쉬운 형태로 서비스가 제공되어 별다른 어려움 없이 사용을 시작할 수 있었고 생각했던 동작을 빠르게 구현해볼 수 있어 상대적으로 정말 빠르고 고생 없이 제작할 수 있었습니다.

그래서 만든 것은...

그간 개인적으로 보고 싶던 데이터가 있었고 볼 수만 있다면 즉시 ML 모델을 만들어 예측 혹은 이상치 감지 등으로 활용할 수 있을 것으로 생각해온 것이 있었습니다. 해당 데이터가 서비스 제공 업체로부터 API로 제공되는 것을 확인했으나 반환하는 JSON의 형태가 난해하여 전처리가 필수인 데다가 저장 공간 또한 상당한 요구사항 일 것이라 가지고 싶다는 생각만 해왔던 것이었습니다.

허나 Github Action의 여러 예시들을 보니 저의 요구사항을 만족시키는 동작이 가능한 것을 확인하였고..

API로부터 데이터를 가져와 빅쿼리에 저장하는 Workflow를 작성하게 되었습니다.

이렇게 쓰라고 만든 물건이 아닌 것 같은 느낌에 조금.. 망설여지지만..

프로세스

작성된 Workflow의 전체적인 흐름은 아래와 같습니다.

1. 최신 데이터를 얻기 위해 가장 마지막 ID를 외부에서 가져와

2. 설정된 Timeout 시간이 되기 전까지,

2.1. API 요청

2.2. 반환받은 JSON 객체를 저장에 용이한 형태로 가공

2.3. 작업 인스턴스의 저장소에 JSONL 파일 형태로 임시 저장

3. 저장된 JSONL 파일을 GCS로 업로드

4. GCS의 JSONL 파일을 Bigquery Table로 삽입

5. 일정 시간 이후 1번부터 재시작.

동시 실행 동작 등의 Github Action Limit을 신경 쓰지 않아도 되었는데,

API가 하나의 요청에 다음 요청의 키가 알려지는 형태로 결과가 반환되는 형식 + 잦은 요청은 거절하는 사용 제한 때문에 애초에 API 제공자에게 요청을 동시에 할 수 없고 + 일부 전처리를 제외한 나머지 동작 모두 GCP에서 동작하는 것이 대부분이기 때문에 Workflow의 사용 제한 가까이 갈 일이 없었습니다.

따라서 요구사항이 매우 단순했고 복잡한 예외처리나 설계가 필요 없이 위의 흐름 정도로 모든 동작이 처리됩니다.

코드

Directory 구조

┬─ .github
│    └ workflows
│        └ collect_stashes.yml : workflow 설정이 들어있는 곳
│
├─ poe
│    ├ get_current_id.py : 가장 마지막 stash id를 얻어오기 위한 함수
│    ├ get_stash.py : API 요청과 전처리, 저장 과정
│    ├ main.py : 전체 프로세스 제어
│    └ set_data.py : 저장된 JSONL을 빅쿼리로 저장하기 위한 과정
│
└─ requirements.txt : 필요한 패키지들 기재, 주로 GCP 연결을 위한 패키지들

작성된 코드는 크게 Workflow를 정의한 yml 파일과 실제 동작을 담당하는 python 파일들로 나뉩니다.

workflows/collect_stashes.yml

Workflow 실행에 가장 중요한 문서입니다. 문법은 공식 문서 Workflow syntax for GitHub Actions를 참조하여 작성되었습니다. 제작한 Workflow에서 주요한 점은,

1. 이벤트 트리거 설정을 cron으로 지정하여 매시간 한 번으로 설정한 것.

on:
schedule:
- cron: '0 * * * *'

2. python-version을 작업 환경과 동일한 3.7로 지정한 것.

    - name: Set up Python
      uses: actions/setup-python@v2
      with:
         python-version: 3.7

3. Action이 실행될 환경에 필요한 dependency 설치

    - name: Install dependencies
      run: |
        python -m pip install --upgrade pip
        if [ -f requirements.txt ]; then pip install -r requirements.txt; fi

4. 실제 작업에서 사용할 코드와 작업 디렉터리 지정

    - name: Run main.py
      run: python main.py
      working-directory: ./poe

5. 코드에 전달할 환경변수 지정

      env:
        GCP_KEY: ${{ secrets.GCP_KEY }}
        GCS_BUCKET: ${{ secrets.GCS_BUCKET }}
        GCP_PROJECT_ID: ${{ secrets.GCP_PROJECT_ID }}

위와 같은 형태로 설정되어 있으며 간단한 설정/문법으로 설정을 마무리할 수 있어 편리했습니다.

Secrets

yml 마지막에 설정해두었던 정보는 모두 Github Secret으로 지정해두었습니다.

공개하고 싶지 않은 정보이고 프로젝트와 버킷 모두 쉽게 변경하고 싶었기 때문입니다.

추후에 파일을 임시로 퍼블릭 조회 권한으로 조정하고 싶은 경우, 버킷의 저장용량이 지나치게 많아질 경우

무료 크레딧을 받을 수 있는 계정으로 교체할 것을 염두에 둔 설정입니다.

poe/set_data.py

GCP와 연계한 주요한 동작이 들어있습니다.

(다른 파이썬 파일들에는 별다른 내용이 들어있지 않아 설명을 생략합니다.)

1. 사용자 인증 정보 등록을 yml에 선언해둔 Github Secret으로 가져와 사용한 것.

def get_gcp_credentials():
    key = os.environ['GCP_KEY']
    credentials = service_account.Credentials.from_service_account_info(json.loads(key, strict=False))
    return credentials

2. GCS로 파일 업로드

def to_gcs(file_name):
    blob_name = 'poe/' + file_name
    bucket = self.gcs.bucket(os.environ['GCS_BUCKET'])
    blob = bucket.blob(blob_name)
    blob.upload_from_filename(f'{TEMP_FILE_DIR}/{file_name}')

3. GCS의 파일을 사용해 Bigquery 테이블 생성

def to_bigquery():
    dataset = self.bq.create_dataset(dataset_id, exists_ok=True)
    table = bigquery.Table(dataset.table(table_id))
    config = bigquery.LoadJobConfig()
    config.write_disposition = 'WRITE_TRUNCATE'
    config.source_format = bigquery.SourceFormat.NEWLINE_DELIMITED_JSON
    config.schema = [
        bigquery.SchemaField('id', 'STRING'),
        ....
        ...
    ]
    config.ignore_unknown_values = True
    config.external_data_configuration = config
    job = self.bq.load_table_from_uri(
        f"gs://{os.environ['GCS_BUCKET']}/poe/*",
        table,
        job_config=config
    )
    job.result()

3.1 NEWLINE_DELIMITED_JSON

- API에서 받아온 형식이 굉장히 복잡한 형태의 JSON 파일이므로 거의 그대로 전달합니다.

- 복잡한 구조의 JSON 형태를 빅쿼리에 넣기 위해 사전에 파일 형태를 파악해두었고,

- config.schema = [...] < 구문을 통해 스키마를 모두 지정하여 전달해주었습니다.

- autodetect = True < 이런 자동 스키마 감지 기능을 사용할 수도 있지만, 복잡한 형태의 데이터는 상당수의 데이터가 누락되거나 지나치게 많은 에러를 뱉어내는 경우가 다반수입니다.

3.2 WRITE_TRUNCATE

- 기존 테이블에 데이터를 추가하는 것이 아닌 모두 지우고 다시 쓰도록 했습니다.

- 데이터 적재 시 기존의 테이블과 새로운 데이터 간 중복 데이터가 있을 가능성 때문에 지난 기록을 지우고 새로 쓰는 과정, 인덱싱을 해두거나 여러 번의 DML 실행을 하는 등의 복잡한 과정을 만들고 싶지 않았습니다.

- 깔끔하게 모두 날려버리고 한 번에 적재하여 그러한 고민을 아예 하지 않는 단순하고 직관적인 동작이길 원했습니다. 추후 데이터 크기에 따라 파티션 컬럼을 지정하거나 테이블 이름에 YYYYMMDD 형식을 추가하여 테이블을 나눠 저장할 것이므로, 이러한 동작으로 구성해놓는 것이 손도 덜 가고 효율적인 구성이 될 것입니다.

- 일반적인 RMDBS가 아닌 Bigquery에서의 동작이므로 이러한 과정이 비효율이나 부하를 일으키지 않는다는 것을 전제로 하는 설계입니다.

준비물은 여기까지, 이제 실행 과정을 지켜보면 됩니다.