RTB house API 데이터 자동화 수집 및 빅쿼리 적재

1. 사전 준비사항

1-1 google cloud 프로젝트 및 빅쿼리 설정

프로젝트 아이디, 데이터 셋을 만들어 줍니다.! 테이블은 자동화로 만들것이라 필요 없습니다.

1-2 API 인증 정보를 얻기!

매체별로 필요한 정보가 다르기 때문에 공식문서를 읽고 찾아야 합니다.
예를 들어 제가 진행한 네이버 검색광고는 api key, customer id, secret key가 필요했고, rtb house는 hash 값과 계정 정보가 필요했어요.

1-3 개발 환경 설정

'빅쿼리 라이브러리 설치 및 구글 클라우드 로그인'

pip install requests google-cloud-bigquery

gcloud auth application-default login

이제 코드를 작성하면 api에서 가져올 수 있습니다.
rtb house에서 사용한 코드 예제입니다.

import requests

# 광고주 해시값 (광고주 ID)
CUSTOMER_ID = ""

# RTBHouse API 인증 정보
API_USERNAME = ""
API_PASSWORD = ""

# 광고 성과 데이터 엔드포인트
API_URL = f"https://api.panel.rtbhouse.com/v5/advertisers/{CUSTOMER_ID}/rtb-stats"

# 요청 파라미터
params = {
    "dayFrom": "2025-01-01",    # 시작 날짜
    "dayTo": "2025-02-17",      # 종료 날짜
    "groupBy": "day",           # 데이터를 일별로 그룹화
    "metrics": "impsCount-clicksCount-campaignCost"  # 필요한 데이터: 노출수, 클릭수, 비용
}

def fetch_rtbhouse_data():
    response = requests.get(API_URL, auth=(API_USERNAME, API_PASSWORD), params=params)
    if response.status_code == 200:
        return response.json().get("data", [])
    else:
        print(f"API 요청 실패: {response.status_code}")
        print(response.text)
        return []

# 테스트: RTBHouse 데이터 확인
data = fetch_rtbhouse_data()
if data:
    print("RTBHouse 데이터 수집 성공!")
    print(data)
else:
    print("데이터가 없습니다.")

2. 빅쿼리에 적재하기

json파일로 형식을 정해놓고 하거나 수동으로 구글 클라우드 테이블, 스키마를 지정해 주는 방법도 있지만, 자동으로 테이블을 만들고 적재하는 함수를 코드에서 짜는게 더 편하다.!

2-1 테이블 자동 생성함수

import time
from google.cloud import bigquery
from google.cloud.exceptions import NotFound

def ensure_table_exists(client, table_ref):
    try:
        table = client.get_table(table_ref)
        print("테이블이 이미 존재합니다.")
        return table
    except NotFound:
        print("테이블이 존재하지 않습니다. 테이블을 생성합니다...")
        schema = [
            bigquery.SchemaField("Date", "STRING", mode="NULLABLE"),
            bigquery.SchemaField("Impressions", "INTEGER", mode="NULLABLE"),
            bigquery.SchemaField("Clicks", "INTEGER", mode="NULLABLE"),
            bigquery.SchemaField("Cost", "FLOAT", mode="NULLABLE"),
        ]
        table = bigquery.Table(table_ref, schema=schema)
        table = client.create_table(table)
        print(f"테이블 생성 완료: {table.full_table_id}")
        return table

2-2 받았던 데이터를 빅쿼리로 전송하는 함수

def send_to_bigquery(data):
    client = bigquery.Client(project=PROJECT_ID)
    table_ref = f"{PROJECT_ID}.{DATASET_ID}.{TABLE_ID}"

    # 테이블 존재 여부 확인 및 생성
    table = ensure_table_exists(client, table_ref)
    
    # 테이블 생성 후 BigQuery에 반영되도록 딜레이 (15초 정도)
    print("테이블 생성 후 반영될 때까지 잠시 대기합니다...")
    time.sleep(15)

    # 데이터를 BigQuery에 적합한 형식으로 변환
    rows_to_insert = [
        {
            "Date": item["day"],
            "Impressions": int(item["impsCount"]),
            "Clicks": int(item["clicksCount"]),
            "Cost": float(item["campaignCost"])
        }
        for item in data
    ]
    
    # 데이터 삽입
    errors = client.insert_rows_json(table_ref, rows_to_insert)
    if errors:
        print(f"데이터 전송 오류: {errors}")
    else:
        print(f"BigQuery에 데이터 전송 완료! {len(rows_to_insert)} rows inserted.")