使用 Google Cloud Platform 為產品設計數據追蹤機制，以 Medium 為例
這篇文章想跟你分享：
開發完後 App 後，往往需要觀察使用者數據以利後續調整 App 的功能。說穿了其實就是一連串資料管線，從收集使用者資料到分析資料的過程，但實際上該如何做呢？本文以 Medium 文章數據為範例，透過網路爬蟲抓取 Medium 上觀看數、完讀數、鼓掌數，在 Google Cloud Platform 建立 Data Pipeline 到 Google BigQuery 以利分析每篇文章的成長趨勢。
本文介紹如何為產品搭建一個數據追蹤的機制，流程包含幾種步驟：
- 收集資料
- 串接原始資料到儲存空間
- 資料清洗
- 由乾淨資料建立成資料表
- 資料應用
為了方便解釋搭建數據追蹤的流程，本文範例以Medium 上的文章為產品，收集文章觀看數、完讀數、鼓掌數為例，建立一個簡單的資料管線，以利後續分析每篇文章的熱度。本文內容分為兩大部分：
- 收集 Medium 數據資料
- 將服務部署至 Google Cloud Platform
Medium 資料放在哪裡？
根據 Medium 官方的 API 文件，目前只開放取得以下幾種資源：
- 使用者資訊：
GET https://api.medium.com/v1/me - 出版刊物：
GET https://api.medium.com/v1/users/{{userId}}/publications - 發表文章：
POST https://api.medium.com/v1/publications/{{publicationId}}/posts - 上傳照片：
POST https://api.medium.com/v1/images
是的你沒看錯 😱，關於各種數據資訊目前無法透過 API 取得，只能透過網站後台的 Dashboard 才能看到。自己的數據自己救，既然官方 API 沒有提供資料，那就自己寫程式來撈取觀看數、閱讀數、鼓掌數吧！
撰寫程式抓取 Medium 後台數據吧！
抓取後台數據的方式百百種，本文提供兩種方式撈取數據：
Method 1. 利用瀏覽器工具查 Medium 網站實際呼叫的內部 API
既然 Medium 只開放在官網後台查看使用數據，那就到後台按下 F12 開啟瀏覽器的 Developer Tool，再到 Network 的標籤底下查詢 Medium 內部呼叫了哪些 API 取得資訊，如下圖所示。
從圖中可以看到 Medium 前端的資訊實際上是呼叫了 Request URL ，還需要附上一些 cookie 才能通過驗證，否則只會收到 401 Unauthorized 。經過反覆實驗後，發現必要的 cookie 為 uid 以及 sid 。
下方範例程式利用 Python 中的 requests 函式庫，簡單對 Request URL 觸發 HTTP GET 請求。收到 HTTP 回應後，還需要根據不同的 URL 收到的網站資料解析網站結構，取得每一篇文章的 views, reads, claps 的數值，範例就不細寫 Parsing 的過程，讀者有興趣可以使用正規表達式取出所需資訊。
import requests
import os
from dotenv import load_dotenv
load_dotenv()
def fetch_data():
# Define all needed variables
URL = os.getenv('URL')
UID = os.getenv('UID')
SID = os.getenv('SID')
cookies = {
'uid': UID,
'sid': SID}
# Send HTTP GET request
stats_page = requests.get(URL, cookies=cookies).text
# Parse stats page to get stats of views and reads
......
......Method 2. 利用 End-to-End 測試工具，模擬使用者登入 Medium 數據後台
另一種做法是使用 End-to-End 測試，例如 Selenium、Playwright、Cypress 等工具，模擬使用者操作登入 Medium 後台 Dashboard。如下方程式碼所示，本文參考 Github 上的爬蟲工具，在 Python 上使用 Selenium 登入取得數據。
下方範例程式可分為四步驟：
- 取得已存在 BigQuery 上的資料
- 使用 Selenium 爬取 Medium 後台數據，再與已儲存在 BigQuery 資料比對 Unique ID，避免程式重複執行時影響 BigQuery 上的資料
- （Optional）如果原始資料無法滿足分析需求，可在此階段進行資料轉換
- 將處理完的資料寫入 BigQuery
透過這四個步驟，可以簡單實作 Data Pipeline 的 ETL，對應關係分別為：
- 1：Extract Data
- 2, 3：Transform Data
- 4：Load Data
def fetch_data(event, context):
# Import libraries
from datetime import datetime, date
from dotenv import load_dotenv
import json
import os
import pandas as pd
from pandas.io import gbq
import pandas_gbq
from medium_stats.scraper import StatGrabberUser
# Define variables and load environment
load_dotenv()
USERNAME = os.getenv('USERNAME')
UID = os.getenv('UID')
SID = os.getenv('SID')
PROJECT_NAME = 'medium-stats-data-flow'
DATASET_NAME = 'medium_data'
TABLE_NAME = 'stats_snapshot'
final_result = {}
# Read exsisted data on BigQuery
df_bq = pd.read_gbq(f"SELECT * FROM `{DATASET_NAME}.{TABLE_NAME}` WHERE DATE(dt) = CURRENT_DATE", project_id=PROJECT_NAME)
existed_article_date = set(df_bq[['post_id', 'dt']].apply('-'.join, axis=1))
# Get summarized stats (data fetehced from https://medium.com/me/stats)
print('Start fetching data from Medium.com')
me = StatGrabberUser(USERNAME, sid=SID, uid=UID, start=datetime.today(), stop=datetime.today())
data = me.get_summary_stats()
# Parse response data into table columns
for item in data:
key = '-'.join([item['postId'], str(date.today())])
if key not in existed_article_date:
final_result[key] = {
'post_id': item['postId'],
'title': item['title'],
'views': item['views'],
'reads': item['reads'],
'claps': item['claps'],
'publish_time': str(item['firstPublishedAt']),
'dt': str(date.today())
}
else:
print(f'The article {key} already existed!')
print('Data fetched done!')
if any(final_result):
# Transform dict into dataframe
df = pd.DataFrame.from_dict(final_result).transpose()
# Write stats data to BigQuery table
print('Start pushing to BigQuery')
df.to_gbq(destination_table=f'{DATASET_NAME}.{TABLE_NAME}', project_id=PROJECT_NAME, if_exists='append')
print('ALL DONE')
部署到雲上吧！
本文利用 Google Cloud Platform 搭建一個簡單的 Data Pipeline，將 Medium 上的觀看數、閱讀數、鼓掌數資料儲存到 BigQuery，以利後續進行文章成長趨勢的分析。整體架構圖如下：
接下來就一一介紹搭建資料管線需要哪些 Google Cloud 上的服務吧！
Step 1. Cloud Pub/Sub 觸發事件推播
寫完資料管線的 ETL 程式後，接下來要做的就是部署到 Google Cloud Platform（文後簡稱 GCP）上，但在部署 Cloud Function 之前需要先設定好「程式執行的時機」、「程式透過什麼訊號觸發」。參考官方文件的說明，Cloud Pub/Sub 是 GCP 上用來傳送事件訊號到各個微服務（Microservice）的機制。想像我們在 GCP 上有 100 個 Microservices，包含 API、虛擬機、Function 等，如果想讓服務們互相快速溝通、避免非同步問題的話，就需要使用 Cloud Pub/Sub 的機制管理訊息的傳遞。如下圖所示，可以直接在主控台建立主題，快速設定 Pub/Sub。
Step 2. Cloud Scheduler 規劃觸發時機
前面提到在部署 Cloud Function 之前需要事先設定好「程式執行的時機」，因此可以使用 GCP 上的排程工具設定週期性執行的時間。如下左圖所示，本文範例設定每日爬取一次資料即可，加上 Medium 更新數據的時間為 GMT+0，因此頻率可設定為每天早上 9 點執行一次程式。
如下右圖所示，如果欲執行的服務不在 GCP 上的話，觸發方式除了 Pub/Sub 以外，還可以使用 HTTP Request 設定好任務排程驅動其他地方的服務。
Step 3. 部署至 Cloud Function 執行任務
設定好 Cloud Pub/Sub 以及 Cloud Scheduler 之，接著就是將資料處理程式部署到雲端上。考量到本文範例程式的性質偏向定期執行一次輕量腳本即可，所以執行任務的服務可以排除掉適合長期 Host Service 的 App Engine、Cloud Run 以及啟動虛擬機建構服務用的 Compute Engine，詳細可以參考 Google 官方的選用指南。
如下圖左所示，本文範例的觸發條件使用 Cloud Pub/Sub，並選取前面步驟建立的主題。如果想從其他非 GCP 的服務呼叫 Cloud Function 也可以使用 HTTP，或者也可以設定當 Cloud Storage / Firestore 的資料更新時執行 Function。
如下圖右所示，在更多設定處也可以自行定義 Runtime 的環境變數，將機敏資訊例如帳號密碼、金鑰等利用環境變數傳入程式碼，避免重要資訊外洩。
設定完成後按下一步，接著就可以將撰寫好的程式碼貼到編輯器中，並且指定進入點為此函式的名稱。Dependancy 的部分也需要一併貼到文件中，函式部署時才會安裝程式需要的套件。套件管理使用 pip 的話可以透過 pip freeze > requirements.txt ，或者 poetry 的話可以利用 poetry export -f requirements.txt -o requirements.txt --without-hashes ，直接將內容貼到 Cloud Function 上。
Step 4. BigQuery 建立查詢資料
如下圖所示，本文範例所建立的資料表僅紀錄每篇文章每日 views、reads、claps 的數值，如果需要進行深入分析的話可以再多增加一些欲觀察的欄位。另外建立資料表時也建議設定資料分區，避免後續資料規模成長後，造成 Query 的成本提高以及效率下降。
如下圖所示，執行完 Cloud Function 後，便可在 BigQuery 上查詢到資料囉（撒花🌼～另外如果需要將資料表再轉換成客製化維度的 Data Warehouse，還可以在 BigQuery 上設定排程，週期性轉換資料表。接下來等待每日 Schedule 跑完後就可以在 BigQuery 上觀察各種指標，例如每日的活躍用戶觀看數量、閱讀數量、用戶活動趨勢，之後可能會再寫一篇分享如何在 BigQuery 上分析各種指標，敬請期待！
總結
雖然目前市面上已經有許多數據追蹤的產品，例如 Google Analytics、AppsFlyer、Amplitude，但對於剛開始起步的服務來說，價格往往是最需要考量的地方。剛好 Google 有提供免費的使用額度，可以提供產品一個數據觀察的 MVP（Minimun Viable Product），在服務規模擴大之前先簡單收集資訊。其實 Data Pipeline 的處理方式大同小異，如果需要再降低成本的話，甚至可以用 Google 雲端硬碟儲存原始資料，再將資料經過 ETL 轉換到 Google Sheet 進行各種指標分析。
如果對於本篇文章有任何想法的話，歡迎直接在下面留言，或者追蹤我們的粉絲專頁 ✿ AppDev Ooops | Facebook ✿ 來跟我們互動唷！謝謝大家！
