许多 Looker 客户都希望让用户不仅能够报告数据仓库中的数据，还能够实际写回和更新该数据仓库。
Looker 通过其 Action API 支持任何数据仓库或目标的这种使用场景。本文档页面将引导使用 Google Cloud 基础架构的客户在 Cloud Run functions 上部署解决方案，以写回 BigQuery。本页面涵盖以下主题：
解决方案注意事项
请使用以下注意事项列表来验证此解决方案是否符合您的需求。
- Cloud Run functions
- 为何选择 Cloud Run functions？ Cloud Run functions 是 Google 的“无服务器”产品，非常适合简化运营和维护。需要注意的一点是，延迟时间（尤其是冷调用）可能比依赖专用服务器的解决方案长。
- 语言和运行时 Cloud Run functions 支持多种语言和运行时。本文档页面将重点介绍 JavaScript 和 Node.js 中的示例。不过，这些概念可以直接转换为其他受支持的语言和运行时。
- BigQuery
- 为何选择 BigQuery？ 虽然本文档页面假定您已在使用 BigQuery，但 BigQuery 通常是数据仓库的绝佳选择。请注意以下几点：
- BigQuery Storage Write API： BigQuery 提供了多种接口来更新数据仓库中的数据，例如基于 SQL 的作业中的数据操纵语言 (DML) 语句。不过，对于大批量写入，最佳选择是 BigQuery Storage Write API。
- 追加而非更新： 即使此解决方案只会追加行，而不会更新行，您也可以始终在查询时从仅追加日志中派生出“当前状态”表，从而模拟更新。
- 为何选择 BigQuery？ 虽然本文档页面假定您已在使用 BigQuery，但 BigQuery 通常是数据仓库的绝佳选择。请注意以下几点：
- 支持服务
- Secret Manager: Secret Manager 会保存密钥值，以确保它们不会存储在过于容易访问的位置，例如直接存储在函数的配置中。
- Identity and Access Management (IAM)： IAM 授权函数在运行时访问必要的密钥，并写入目标 BigQuery 表。
- Cloud Build： 虽然本页面不会深入讨论 Cloud Build，但 Cloud Run functions 会在后台使用它，您可以使用 Cloud Build 自动将对 Git 代码库中源代码的更改持续部署到函数。
- 操作和用户身份验证
- Cloud Run 服务账号 将 Looker 操作与组织自己的第一方资产和资源集成的主要且最简单的方法是，使用 Looker Action API 的基于令牌的身份验证机制对来自 Looker 实例的请求进行身份验证，然后使用服务账号授权函数更新 BigQuery 中的数据。
- OAuth： 另一种选择（本页面未介绍）是使用 Looker Action API 的 OAuth 功能。这种方法更复杂，通常不需要，但如果您需要使用 IAM 定义最终用户写入表的权限，而不是使用他们在 Looker 中的权限或函数代码中的临时逻辑，则可以使用此方法。
演示代码演示
我们在 GitHub 上提供了一个文件，其中包含演示操作的全部逻辑 可供使用。在本部分中，我们将介绍代码的关键元素。
设置代码
第一部分包含一些演示常量，用于标识操作将写入的表。在本页面后面的部署指南部分中，系统会指示您将项目 ID 替换为您自己的项目 ID，这将是对代码的唯一必要修改。
/*** Demo constants */
const projectId = "your-project-id"
const datasetId = "demo_dataset"
const tableId = "demo_table"
下一部分声明并初始化操作将使用的一些代码依赖项。我们提供了一个示例，该示例使用 Secret Manager Node.js 模块“在代码中”访问 Secret Manager；不过，您也可以使用 Cloud Run functions 的内置功能在初始化期间为您检索密钥，从而消除此代码依赖项。
/*** Code Dependencies ***/
const crypto = require("crypto")
const {SecretManagerServiceClient} = require('@google-cloud/secret-manager')
const secrets = new SecretManagerServiceClient()
const BigqueryStorage = require('@google-cloud/bigquery-storage')
const BQSManagedWriter = BigqueryStorage.managedwriter
请注意，引用的 @google-cloud 依赖项也在我们的 package.json 文件中声明，以允许预加载这些依赖项并使其可供我们的 Node.js 运行时使用。crypto 是一个内置的 Node.js 模块，未在 package.json 中声明。
HTTP 请求处理和路由
您的代码向 Cloud Run functions 运行时公开的主要接口是一个导出的 JavaScript 函数，该函数遵循 Node.js Express Web 服务器惯例。具体而言，您的函数会接收两个实参：第一个实参表示 HTTP 请求，您可以从中读取各种请求参数和值；第二个实参表示响应对象，您将响应数据发送到该对象。虽然函数的名称可以是您想要的任何名称，但您必须稍后向 Cloud Run functions 提供该名称，如部署指南部分中所述。
/*** Entry-point for requests ***/
exports.httpHandler = async function httpHandler(req,res) {
httpHandler 函数的第一部分声明了我们的操作将识别的各种路由，这些路由与单个操作所需的 Action API 端点以及将处理每个路由的函数（稍后在文件中定义）紧密对应。
虽然某些操作 + Cloud Run functions 示例会为每个此类路由部署一个单独的函数，以与 Cloud Run functions 的默认路由一一对应，但函数能够在代码中应用额外的“子路由”，如这里所示。这最终取决于个人偏好，但在代码中执行此额外的路由可以最大限度地减少我们必须部署的函数数量，并帮助我们在所有操作的端点中保持单一连贯的代码状态。
const routes = {
"/": [hubListing],
"/status": [hubStatus], // Debugging endpoint. Not required.
"/action-0/form": [
requireInstanceAuth,
action0Form
],
"/action-0/execute": [
requireInstanceAuth,
processRequestBody,
action0Execute
]
}
HTTP 处理程序函数的其余部分实现了针对上述路由声明的 HTTP 请求的处理，并将这些处理程序的返回值连接到响应对象。
try {
const routeHandlerSequence = routes[req.path] || [routeNotFound]
for(let handler of routeHandlerSequence) {
let handlerResponse = await handler(req)
if (!handlerResponse) continue
return res
.status(handlerResponse.status || 200)
.json(handlerResponse.body || handlerResponse)
}
}
catch(err) {
console.error(err)
res.status(500).json("Unhandled error. See logs for details.")
}
}
在 HTTP 处理程序和路由声明完成后，我们将深入探讨必须实现的三个主要操作端点：
操作列表端点
当 Looker 管理员首次将 Looker 实例连接到操作服务器时，Looker 将调用提供的网址（称为“操作列表端点”），以获取有关可通过服务器执行的操作的信息。
在我们之前展示的路由声明中，我们在函数的网址下的根路径 (/) 中提供了此端点，并指明它将由 hubListing 函数处理。
从以下函数定义中可以看出，它根本没有太多“代码”，它只是每次都返回相同的 JSON 数据。需要注意的一点是，它确实会动态地将其“自己的”网址包含到某些字段中，从而允许 Looker 实例将后续请求发送回同一函数。
async function hubListing(req){
return {
integrations: [
{
name: "demo-bq-insert",
label: "Demo BigQuery Insert",
supported_action_types: ["cell", "query", "dashboard"],
form_url:`${process.env.CALLBACK_URL_PREFIX}/action-0/form`,
url: `${process.env.CALLBACK_URL_PREFIX}/action-0/execute`,
icon_data_uri: "data:image/png;base64,...",
supported_formats:["inline_json"],
supported_formattings:["unformatted"],
required_fields:[
// You can use this to make your action available
// for specific queries/fields
// {tag:"user_id"}
],
params: [
// You can use this to require parameters, either
// from the Action's administrative configuration,
// or from the invoking user's user attributes.
// A common use case might be to have the Looker
// instance pass along the user's identification to
// allow you to conditionally authorize the action:
{name: "email", label: "Email", user_attribute_name: "email", required: true}
]
}
]
}
}
出于演示目的，我们的代码不需要进行身份验证即可检索此列表。不过，如果您认为自己的操作元数据是敏感信息，也可以要求对此路由进行身份验证，如下一部分所示。
另请注意，我们的 Cloud Run 函数可以公开和处理多个操作，这解释了我们的路由惯例 /action-X/...。不过，我们的演示 Cloud Run 函数将仅实现一个操作。
操作表单端点
虽然并非所有使用场景都需要表单，但拥有表单非常适合数据库写回的使用场景，因为用户可以在 Looker 中检查数据，然后提供要插入到数据库中的值。由于我们的操作列表提供了 form_url 参数，因此当用户开始与您的操作互动时，Looker 将调用此 操作表单端点，以确定要从用户那里捕获哪些额外数据。
在我们的路由声明中，我们在 /action-0/form 路径下提供了此端点，并将其与两个处理程序相关联：requireInstanceAuth 和 action0Form。
我们设置了这样的路由声明，以允许使用多个处理程序，因为某些逻辑可以重复用于多个端点。
例如，我们可以看到 requireInstanceAuth 用于多个路由。我们会在需要请求必须来自 Looker 实例的任何地方使用此处理程序。该处理程序会从 Secret Manager 检索预期的令牌值，并拒绝任何没有该预期令牌值的请求。
async function requireInstanceAuth(req) {
const lookerSecret = await getLookerSecret()
if(!lookerSecret){return}
const expectedAuthHeader = `Token token="${lookerSecret}"`
if(!timingSafeEqual(req.headers.authorization,expectedAuthHeader)){
return {
status:401,
body: {error: "Looker instance authentication is required"}
}
}
return
function timingSafeEqual(a, b) {
if(typeof a !== "string"){return}
if(typeof b !== "string"){return}
var aLen = Buffer.byteLength(a)
var bLen = Buffer.byteLength(b)
const bufA = Buffer.allocUnsafe(aLen)
bufA.write(a)
const bufB = Buffer.allocUnsafe(aLen) //Yes, aLen
bufB.write(b)
return crypto.timingSafeEqual(bufA, bufB) && aLen === bLen;
}
}
请注意，我们使用的是 timingSafeEqual 实现，而不是标准等式检查 (==)，以防止泄露侧信道时序信息，从而让攻击者能够快速找出密钥的值。
假设请求通过了实例身份验证检查，则该请求将由 action0Form 处理程序处理。
async function action0Form(req){
return [
{name: "choice", label: "Choose", type:"select", options:[
{name:"Yes", label:"Yes"},
{name:"No", label:"No"},
{name:"Maybe", label:"Maybe"}
]},
{name: "note", label: "Note", type: "textarea"}
]
}
虽然我们的演示示例非常静态，但表单代码可能更具互动性对于某些使用场景来说。例如，根据用户在初始下拉菜单中的选择，可以显示不同的字段。
操作执行端点
操作执行端点是任何操作的大部分逻辑所在的位置，也是我们将介绍特定于 BigQuery 插入使用场景的逻辑的位置。
在我们的路由声明中，我们在 /action-0/execute 路径下提供了此端点，并将其与三个处理程序相关联：requireInstanceAuth、processRequestBody 和 action0Execute。
我们已经介绍了 requireInstanceAuth，而 processRequestBody 处理程序主要提供一些不怎么有趣的预处理，以将 Looker 请求正文中的某些不方便的字段转换为更方便的格式，但您可以在完整代码文件中参考它。
action0Execute 函数首先展示了从操作请求的几个部分提取信息的示例，这些信息可能很有用。实际上，请注意，我们的代码称为 formParams 和 actionParams 的请求元素可以包含不同的字段，具体取决于您在列表和表单端点中声明的内容。
async function action0Execute (req){
try{
// Prepare some data that we will insert
const scheduledPlanId = req.body.scheduled_plan && req.body.scheduled_plan.scheduled_plan_id
const formParams = req.body.form_params || {}
const actionParams = req.body.data || {}
const queryData = req.body.attachment.data //If using a standard "push" action
/*In case any fields require datatype-specific preparation, check this example:
https://github.com/googleapis/nodejs-bigquery-storage/blob/main/samples/append_rows_proto2.js
*/
const newRow = {
invoked_at: new Date(),
invoked_by: actionParams.email,
scheduled_plan_id: scheduledPlanId || null,
query_result_size: queryData.length,
choice: formParams.choice,
note: formParams.note,
}
然后，代码会转换为一些标准 BigQuery 代码，以实际插入数据。请注意，BigQuery Storage Write API 提供了其他更复杂的变体，更适合持久流式传输连接或批量插入大量记录；但是，对于在 Cloud Run 函数的上下文中响应单个用户互动，这是最直接的变体。
await bigqueryConnectAndAppend(newRow)
...
async function bigqueryConnectAndAppend(row){
let writerClient
try{
const destinationTablePath = `projects/${projectId}/datasets/${datasetId}/tables/${tableId}`
const streamId = `${destinationTablePath}/streams/_default`
writerClient = new BQSManagedWriter.WriterClient({projectId})
const writeMetadata = await writerClient.getWriteStream({
streamId,
view: 'FULL',
})
const protoDescriptor = BigqueryStorage.adapt.convertStorageSchemaToProto2Descriptor(
writeMetadata.tableSchema,
'root'
)
const connection = await writerClient.createStreamConnection({
streamId,
destinationTablePath,
})
const writer = new BQSManagedWriter.JSONWriter({
streamId,
connection,
protoDescriptor,
})
let result
if(row){
// The API expects an array of rows, so wrap the single row in an array
const rowsToAppend = [row]
result = await writer.appendRows(rowsToAppend).getResult()
}
return {
streamId: connection.getStreamId(),
protoDescriptor,
result
}
}
catch (e) {throw e}
finally{
if(writerClient){writerClient.close()}
}
}
演示代码还包含一个“状态”端点，用于问题排查，但 Action API 集成不需要此端点。
部署指南
最后，我们将提供一个分步指南，用于自行部署演示，涵盖前提条件、Cloud Run 函数部署、BigQuery 配置和 Looker 配置。
项目和服务前提条件
在开始配置任何具体内容之前，请查看此列表，了解解决方案需要哪些服务和政策：
- 新项目： 您需要一个新项目来存放示例中的资源。
- 服务： 首次在 Cloud 控制台界面中使用 BigQuery 和 Cloud Run functions 时，系统会提示您为必要服务启用必需的 API，包括 BigQuery、Artifact Registry、Cloud Build、Cloud Functions、Cloud Logging、Pub/Sub、Cloud Run Admin 和 Secret Manager。
- 未经身份验证的调用的政策： 此使用场景要求我们部署“允许公开访问”的 Cloud Run functions，因为我们将根据 Action API 在代码中处理传入请求的身份验证，而不是使用 IAM。虽然默认允许这样做，但组织政策通常会限制此用法。具体而言，
constraints/iam.allowedPolicyMemberDomains政策会限制可以授予 IAM 权限的人员，您可能需要对其进行调整，以允许allUsers主账号进行未经身份验证的访问。如果您发现自己无法允许公开访问，请参阅此指南：如何在实施网域限定共享时创建公共 Cloud Run 服务，了解详情。 - 其他政策： 请注意，其他Google Cloud 组织政策限制条件也可能会阻止部署默认情况下允许的服务。
部署 Cloud Run 函数
创建新项目后，请按照以下步骤部署 Cloud Run 函数
- 在 Cloud Run functions 中，点击 创建函数。
- 为函数选择任意名称（例如“demo-bq-insert-action”）。
- 在触发器 设置下：
- 触发器类型应已为“HTTPS”。
- 将身份验证 设置为允许未经身份验证的调用 。
- 将 网址 值复制到剪贴板。
- 在运行时 > 运行时环境变量 设置下：
- 点击添加变量 。
- 将变量名称设置为
CALLBACK_URL_PREFIX。 - 将上一步中的网址粘贴为值。
- 点击下一步 。
- 点击
package.json文件，然后粘贴 内容。 - 点击
index.js文件，然后粘贴 内容。 - 将文件顶部的
projectId变量分配给您自己的项目 ID。 - 将入口点 设置为
httpHandler。 - 点击部署 。
- 向 build 服务账号授予所请求的权限（如果有）。
- 等待部署完成。
- 如果在任何后续步骤中，您收到一条错误消息，指示您查看 Google Cloud 日志，请注意，您可以从此页面的 日志 标签页访问此函数的日志。
- 在离开 Cloud Run 函数的页面之前，请在详细信息 标签页下找到并记下函数拥有的服务账号 。我们将在后续步骤中使用此账号，以确保函数具有所需的权限。
- 通过访问网址，直接在浏览器中测试函数部署。您应该会看到包含集成列表的 JSON 响应。
- 如果您收到 403 错误，则您尝试设置允许未经身份验证的调用 可能因组织政策而以静默方式失败。检查您的函数是否允许未经身份验证的调用，查看您的组织政策设置，然后尝试更新该设置。
访问 BigQuery 目标表
实际上，要插入的目标表可以位于不同的 Google Cloud 项目中；不过，出于演示目的，我们将在同一项目中创建一个新的目标表。无论哪种情况，您都需要确保 Cloud Run 函数的服务账号具有写入表的权限。
- 前往 BigQuery 控制台。
创建演示表：
- 在探索器栏中，使用项目旁边的省略号菜单，然后选择创建数据集 。
- 为数据集指定 ID
demo_dataset，然后点击创建数据集 。 - 使用新创建的数据集上的省略号菜单，然后选择创建表 。
- 为表指定名称
demo_table。 在 Schema 下，选择 以文本形式修改 ，使用以下架构，然后点击 创建表 。
[ {"name":"invoked_at","type":"TIMESTAMP"}, {"name":"invoked_by","type":"STRING"}, {"name":"scheduled_plan_id","type":"STRING"}, {"name":"query_result_size","type":"INTEGER"}, {"name":"choice","type":"STRING"}, {"name":"note","type":"STRING"} ]
分配权限：
- 在探索器 栏中，点击您的数据集。
- 在数据集 页面上，点击共享 > 权限 。
- 点击添加主账号 。
- 将新主账号 设置为函数的服务账号，如本页面前面所述。
- 分配 BigQuery Data Editor 角色。
- 点击保存 。
连接到 Looker
现在函数已部署，我们将把 Looker 连接到该函数。
- 我们需要一个共享密钥令牌，以便操作对请求是否来自 Looker 实例进行身份验证。生成一个较长的随机字符串并妥善保管。我们将在后续步骤中将其用作 Looker 密钥 值。
- 在 Cloud 控制台中，前往 Secret Manager。
- 点击创建密钥 。
- 将名称 设置为
LOOKER_SECRET。（此名称在此演示的代码中是硬编码的，但在使用自己的代码时，您可以有效地选择任何名称。） - 将密钥值 设置为您生成的密钥值。
- 点击创建密钥 。
- 在密钥 页面上，点击权限 标签页。
- 点击授予访问权限 。
- 将新主账号 设置为函数的服务账号，如前所述。
- 分配 Secret Manager Secret Accessor 角色。
- 点击保存 。
- 您可以通过访问附加到函数网址的
/status路由来确认函数是否成功访问密钥。
- 在 Looker 实例中：
- 依次前往“管理”>“平台”>“操作”。
- 前往页面底部，点击添加操作中心 。
- 提供函数的网址（例如 https://your-region-your-project.cloudfunctions.net/demo-bq-insert-action），然后点击添加操作中心 进行确认。
- 您现在应该会看到一个新的操作中心条目，其中包含一个名为 Demo BigQuery Insert 的操作。
- 在操作中心条目中，点击配置授权 。
- 在授权令牌 字段中输入生成的 Looker 密钥 ，然后点击更新令牌 。
- 在 Demo BigQuery Insert 操作中，点击启用 。
- 将已启用 开关切换为开启。
- 系统应会自动运行操作测试，以确认函数正在接受 Looker 的请求并正确响应表单端点。
- 点击保存 。
端到端测试
我们现在应该能够实际使用新操作了。此操作配置为可与任何查询搭配使用，因此请选择任意探索（例如内置的系统活动探索），向新查询添加一些字段，运行该查询，然后从齿轮菜单中选择发送 。您应该会将该操作视为可用目标之一，并且系统会提示您输入一些字段：
按 Send 后，您应该会在 BigQuery 表中插入一个新行（并在 invoked_by 列中标识 Looker 用户账号的电子邮件地址）！