如何触发数据从 Cloud Storage 上传到 Kubernetes Engine 的 BigQuery？

Posted 2023-03-24

【中文标题】如何触发数据从 Cloud Storage 上传到 Kubernetes Engine 的 BigQuery？

【英文标题】：How to trigger data upload from Cloud Storage to BigQuery from Kubernetes Engine?

【发布时间】：2017-11-27 08:22:43

【问题描述】：

API 服务器正在 Kubernetes Engine (GKE) 上运行。用户可以将相对较小的数据集（约 100mb，具有相同数据结构的多个 .csv）从客户端应用程序上传到云存储 (GCS)。上传完成后，我需要将所有新的 .csv 文件中的所有数据导入到单个现有 BigQuery 表中，其中包含一些特定于用户的参数（用用户 ID 标记每一行可能是左右）。顺序无关紧要。

Google 文档为此提供了基于 GUI 的解决方案和命令行解决方案。虽然，我认为，有一种方法可以触发上传并从基于 GKE 的服务器本身跟踪它的进度。我该怎么做？

不确定这是否重要：GKE api 服务器是在 NodeJS 上编写的。

【问题讨论】：

您能否将 CSV 文件定义为联合表（作为 BigQuery 查询作业的一部分），然后运行类似 SELECT *, <user_id> FROM MyCsvTable; 的查询？您可以指定附加到现有表作为查询作业的一部分。

@ElliottBrossard 您的意思是“从 csv 查询而不将其上传到 BigQuery”？我认为这将比上传后在 BigQuery 本身中处理这些数据效率低得多

这取决于我猜的型号。如果您想一次处理多个 100MB 的文件，那么我同意先将 CSV 文件加载到 BigQuery 是更好的选择。对于“实时”解决方案，您可以使用联合表。无论如何，您是否尝试过使用 NodeJS 中的 BigQuery API？听起来这就是你需要使用的。

我在“来自 NodeJS 的 BigQuery API”中没有找到解决方案

这是文档中的代码示例：cloud.google.com/bigquery/docs/…

【参考方案1】：

这里是上传文件到 GCS 的示例，取自 BigQuery documentation。您可以根据需要配置作业；该页面上有一些参考资料和带有附加功能的link to the GitHub repo：

// Imports the Google Cloud client libraries
const BigQuery = require('@google-cloud/bigquery');
const Storage = require('@google-cloud/storage');

// The project ID to use, e.g. "your-project-id"
// const projectId = "your-project-id";

// The ID of the dataset of the table into which data should be imported, e.g. "my_dataset"
// const datasetId = "my_dataset";

// The ID of the table into which data should be imported, e.g. "my_table"
// const tableId = "my_table";

// The name of the Google Cloud Storage bucket where the file is located, e.g. "my-bucket"
// const bucketName = "my-bucket";

// The name of the file from which data should be imported, e.g. "file.csv"
// const filename = "file.csv";

// Instantiates clients
const bigquery = BigQuery(
  projectId: projectId
);

const storage = Storage(
  projectId: projectId
);

let job;

// Imports data from a Google Cloud Storage file into the table
bigquery
  .dataset(datasetId)
  .table(tableId)
  .import(storage.bucket(bucketName).file(filename))
  .then((results) => 
    job = results[0];
    console.log(`Job $job.id started.`);

    // Wait for the job to finish
    return job.promise();
  )
  .then((results) => 
    // Get the job's status
    return job.getMetadata();
  ).then((metadata) => 
    // Check the job's status for errors
    const errors = metadata[0].status.errors;
    if (errors && errors.length > 0) 
      throw errors;
    
  ).then(() => 
    console.log(`Job $job.id completed.`);
  )
  .catch((err) => 
    console.error('ERROR:', err);
  );

上传后，您可以run a query 查询新上传的 CSV 文件并将结果附加到所需的目标表中。