Google Cloud：使用遠(yuǎn)程和事件驅(qū)動(dòng) AI Platform Pipelines

機(jī)器學(xué)習(xí)工作流可能涉及相互依賴的許多步驟 —— 從數(shù)據(jù)準(zhǔn)備和分析、訓(xùn)練、評(píng)估到部署等。很難以一種特別的方式對(duì)這些流程進(jìn)行組合和跟蹤—— 例如，以一套 notebook 或者腳本 —— 而審核和可重復(fù)性之類的問(wèn)題會(huì)變得越來(lái)越嚴(yán)重。

今年早些時(shí)候推出的Cloud AI Platform Pipelines 有助于解決這些問(wèn)題：AI Platform Pipelines 提供了一種部署健壯、可重復(fù)機(jī)器學(xué)習(xí)流水線以及監(jiān)控、審核、版本跟蹤和可重復(fù)性的方法，并且為您的ML 工作流提供了一個(gè)企業(yè)就緒、易安裝的安全執(zhí)行環(huán)境.

盡管通過(guò) Pipelines DashboardUI 可以簡(jiǎn)便地上傳、運(yùn)行和監(jiān)控流水線，但有時(shí)您可能要以編程方式訪問(wèn) Pipelines 框架。這樣可以讓您從 notebook 構(gòu)建和運(yùn)行流水線，并以編程方式管理您的流水線、試驗(yàn)和運(yùn)行。首先，您需要對(duì)您的Pipelines 安裝端點(diǎn)進(jìn)行身份驗(yàn)證。如何做到這一點(diǎn)取決于您的代碼運(yùn)行所基于的環(huán)境。所以，這是我們現(xiàn)在要關(guān)注的焦點(diǎn)。

事件驅(qū)動(dòng)的Pipeline 調(diào)用

我們將要介紹的一類有趣的用例是通過(guò) Cloud Functions 之類的服務(wù)使用SDK 來(lái)設(shè)置事件驅(qū)動(dòng)的 Pipeline 調(diào)用。這些用例允許您基于添加到GCS 存儲(chǔ)桶的新數(shù)據(jù)、添加到 PubSub 主題的新信息或者其他事件來(lái)開(kāi)始部署。

通過(guò)Pipelines Dashboard UI 可以簡(jiǎn)便地上傳和運(yùn)行流水線，但您也經(jīng)常需要進(jìn)行遠(yuǎn)程訪問(wèn)。

通過(guò) AI Platform Pipelines，您可以使用Kubeflow Pipelines (KFP) SDK 或者通過(guò)采用TFX SDK 自定義TensorFlow Extended(TFX)Pipeline 模板，來(lái)指定流水線。要使用 SDK 從 Pipelines 集群外部進(jìn)行連接，必須在遠(yuǎn)程環(huán)境中設(shè)置您的憑據(jù)，使您有權(quán)訪問(wèn)AI Platform Pipelines 安裝端點(diǎn)。多數(shù)情況下，為您的賬戶安裝和初始化 gcloud 很簡(jiǎn)單（或者就像AI Platform Notebooks 一樣，已經(jīng)為您設(shè)置好了），連接是透明的。

而如果您是運(yùn)行在 Google Cloud 之上，在一種不能簡(jiǎn)單地進(jìn)行 gcloud 初始化的場(chǎng)景下，您可以通過(guò)底層 VM 的元數(shù)據(jù)獲得并使用訪問(wèn)令牌來(lái)進(jìn)行身份驗(yàn)證。如果運(yùn)行時(shí)環(huán)境正在使用與 Pipelines 安裝所使用的不同的服務(wù)賬戶，您還需要為該服務(wù)賬戶提供訪問(wèn) Pipelines 端點(diǎn)的權(quán)限。例如，這種情況就是這樣，采用 Cloud Functions，其實(shí)例使用項(xiàng)目的App Engine 服務(wù)賬戶。

最后，如果您未在 Google Cloud 中運(yùn)行并且未安裝gcloud，可使用服務(wù)賬戶憑據(jù)文件來(lái)生成訪問(wèn)令牌。

接下來(lái)，我們將介紹這些選項(xiàng)，并且提供一個(gè)示例來(lái)說(shuō)明如何定義對(duì)流水線運(yùn)行進(jìn)行初始化的Cloud Function，從而使您能夠設(shè)置事件驅(qū)動(dòng)的 Pipeline 作業(yè)。

使用 Kubeflow PipelinesSDK 通過(guò) gcloud 訪問(wèn)連接 AI Platform Pipelines 集群

要連接 AI PlatformPipelines 集群，您首先需要找到其端點(diǎn)的 URL。

一個(gè)簡(jiǎn)單的方法是訪問(wèn)您的 AIPipelines 儀表板，并點(diǎn)擊“SETTINGS”（設(shè)置）。

點(diǎn)擊“Settings”獲得有關(guān)您的安裝的信息。

將彈出一個(gè)與下圖類似的窗口：

KFP 客戶端設(shè)置

復(fù)制所顯示的代碼段以使用 KFP SDK 連接您的安裝端點(diǎn)。這一簡(jiǎn)單的 notebook 示例旨在讓您對(duì)流程進(jìn)行測(cè)試。（在此另有一個(gè)使用 TFX SDK 和 TFXTemplates 的示例）。

從AI Platform Notebooks 進(jìn)行連接

如果您使用在相同項(xiàng)目中運(yùn)行的AI Platform Notebook，連通性會(huì)起作用。您只需如上所述提供您的Pipelines 安裝端點(diǎn)的 URL。

從本地或者開(kāi)發(fā)機(jī)器進(jìn)行連接

您可能要從您的本地機(jī)器或者其他類似環(huán)境進(jìn)行Pipelines 安裝部署。如果已安裝 gcloud 并為您的賬號(hào)授權(quán)，身份驗(yàn)證應(yīng)當(dāng)同樣直接有效。

從GCP 運(yùn)行時(shí)連接 AI Platform Pipelines 端點(diǎn)

對(duì)于無(wú)服務(wù)器環(huán)境（例如，CloudFunctions、Cloud Run 或者 App Engine），由于有使用不同服務(wù)賬戶的臨時(shí)實(shí)例，設(shè)置和初始化 gcloud 可能存在問(wèn)題。這里我們將使用不同的方法：我們將允許服務(wù)賬戶訪問(wèn)Cloud AI Pipelines 的反向代理，并且獲得創(chuàng)建客戶端對(duì)象時(shí)傳遞的訪問(wèn)令牌。我們將通過(guò)一個(gè) Cloud Functions 示例詳細(xì)說(shuō)明如何這樣做。

示例：使用Cloud Functions 進(jìn)行事件驅(qū)動(dòng)的流水線部署

Cloud Functions 是 Google Cloud 的事件驅(qū)動(dòng)無(wú)服務(wù)器計(jì)算平臺(tái)。使用 Cloud Functions 觸發(fā)流水線部署會(huì)為支持事件驅(qū)動(dòng)的流水線帶來(lái)許多可能性，您可以基于添加到 Google Cloud Storage 存儲(chǔ)桶的新數(shù)據(jù)、添加到 PubSub 主題的新信息等來(lái)開(kāi)始部署。

例如，一旦新的一批數(shù)據(jù)到達(dá)或者 AIPlatform Data Labeling Service“導(dǎo)出”完成，您會(huì)預(yù)期著自動(dòng)開(kāi)始運(yùn)行 ML 訓(xùn)練流水線。

在此，您將看到一個(gè)通過(guò)向 CloudStorage 存儲(chǔ)桶添加新文件觸發(fā)流水線部署的示例。

對(duì)于這種情況，您可能不想在保存您的數(shù)據(jù)集的Cloud Storage 存儲(chǔ)桶中設(shè)置 Cloud Functions 觸發(fā)器，因?yàn)檫@在每次添加文件時(shí)都會(huì)觸發(fā) —— 如果更新包括多個(gè)文件，可能這并非您想要的行為。相反，一旦完成數(shù)據(jù)導(dǎo)出或者攝取流程，您可以將Cloud Storage 文件寫(xiě)入一個(gè)單獨(dú)的“觸發(fā)器桶”，該文件包含有關(guān)新添加的數(shù)據(jù)的路徑信息。通過(guò)定義一個(gè)在該存儲(chǔ)桶上觸發(fā)的 Cloud Functions 函數(shù)，這會(huì)讀取文件內(nèi)容并且在啟動(dòng)流水線運(yùn)行時(shí)將數(shù)據(jù)路徑信息作為參數(shù)進(jìn)行傳遞。

通過(guò)設(shè)置 Cloud Functions 函數(shù)來(lái)部署流水線有兩個(gè)主要步驟。首先是提供由Cloud Functions 使用的服務(wù)賬戶 —— 您的項(xiàng)目的 App Engine 服務(wù)賬戶 —— 通過(guò)將其添加為具有 Project Viewer 權(quán)限的成員，訪問(wèn)由 Pipelines 安裝使用的服務(wù)賬戶。默認(rèn)情況下，Pipelines服務(wù)賬戶將是您的項(xiàng)目的 Compute Engine 默認(rèn)服務(wù)賬戶。

然后，您定義并部署Cloud Functions函數(shù)，該函數(shù)會(huì)觸發(fā)啟動(dòng)流水線運(yùn)行。該函數(shù)獲得 Cloud Functions 實(shí)例的服務(wù)賬戶訪問(wèn)令牌，此令牌被傳遞至KFP 客戶端構(gòu)造函數(shù)。接下來(lái)，您可以通過(guò)客戶端對(duì)象啟動(dòng)流水線運(yùn)行（或者提出其他請(qǐng)求）。

有關(guān)觸發(fā) Cloud Storage 文件或者其內(nèi)容的信息可作為流水線運(yùn)行時(shí)參數(shù)被傳遞。

import kfp

def get_access_token():

  url ='http://metadata.google.internal/computeMetadata/v1/instance/service-accounts/default/token'

   r= requests.get(url, headers={'Metadata-Flavor': 'Google'})

  r.raise_for_status()

  access_token = r.json()['access_token']

  return access_token

...

token = get_access_token()

client = kfp.Client(host=HOST_URL,existing_token=token)

...

res = client.run_pipeline(...)

因?yàn)?Cloud Function 需要安裝 kfp SDK，您需要定義一個(gè) CloudFunctions 部署所使用的 requirements.txt 文件，該文件將指定此要求。

此 notebook會(huì)引導(dǎo)您完成設(shè)置流程，并顯示 CloudFunctions 函數(shù)代碼。此示例定義了一個(gè)非常簡(jiǎn)單的流水線，只回顯一個(gè)作為參數(shù)傳遞的文件名。Cloud Function 啟動(dòng)該流水線的運(yùn)行，傳遞Function 調(diào)用所觸發(fā)的新文件或者被修改文件的文件名。

使用服務(wù)賬戶憑據(jù)文件連接Pipelines 端點(diǎn)

如果您正在進(jìn)行本地開(kāi)發(fā)，并且未安裝 gcloud，也可通過(guò)本地可用的服務(wù)賬戶憑據(jù)文件獲得憑據(jù)令牌。本示例介紹了如何做到這一點(diǎn)。最簡(jiǎn)單的方式是使用與用于Pipelines 安裝相同的服務(wù)賬戶的憑據(jù)  —— 默認(rèn)為 ComputeEngine 服務(wù)賬戶。否則，您需要為其他服務(wù)賬戶提供對(duì) Compute Engine 賬戶的訪問(wèn)權(quán)限。

總結(jié)

您可以采用幾種方法來(lái)使用 AI Platform Pipelines API 遠(yuǎn)程部署流水線，我們?cè)诖私榻B的notebook 應(yīng)當(dāng)是一個(gè)很好的開(kāi)端。尤其是，Cloud Functions 讓您能夠支持許多類型的事件驅(qū)動(dòng)的流水線。要了解有關(guān)將這些知識(shí)付諸實(shí)踐的更多信息，請(qǐng)參考 Cloud Functions notebook 了解關(guān)于如何基于新數(shù)據(jù)自動(dòng)啟動(dòng)流水線運(yùn)行的示例。嘗試運(yùn)行這些 notebook，并讓我們了解您的想法。您可以在Twitter 通過(guò) @amygdala 找到我。