如何使用在 GCP 之外运行的 Spark 访问 BigQuery答案

【问题标题】：How to access BigQuery using Spark which is running outside of GCP如何使用在 GCP 之外运行的 Spark 访问 BigQuery
【发布时间】：2021-03-21 05:27:16
【问题描述】：

我正在尝试将在私有数据中心上运行的 Spark 作业与 BigQuery 连接起来。我已经创建了服务帐户并获得了私有 JSON 密钥，并获得了对我想要查询的数据集的读取权限。但是，当我尝试与 Spark 集成时，我收到了User does not have bigquery.tables.create permission for dataset xxx:yyy.。我们是否需要创建表权限才能使用 BigQuery 从表中读取数据？

以下是在控制台上打印的响应，

{
  "code" : 403,
  "errors" : [ {
    "domain" : "global",
    "message" : "Access Denied: Dataset xxx:yyy: User does not have bigquery.tables.create permission for dataset xxx:yyy.",
    "reason" : "accessDenied"
  } ],
  "message" : "Access Denied: Dataset xxx:yyy: User does not have bigquery.tables.create permission for dataset xxx:yyy.",
  "status" : "PERMISSION_DENIED"
}

下面是我尝试访问 BigQuery 的 Spark 代码

object ConnectionTester extends App {


  val session = SparkSession.builder()
    .appName("big-query-connector")
    .config(getConf)
    .getOrCreate()


    session.read
      .format("bigquery")
      .option("viewsEnabled", true)
    .load("xxx.yyy.table1")
    .select("col1")
    .show(2)


  private def getConf : SparkConf = {
    val sparkConf = new SparkConf
    sparkConf.setAppName("biq-query-connector")
    sparkConf.setMaster("local[*]")
    sparkConf.set("parentProject", "my-gcp-project")
    sparkConf.set("credentialsFile", "<path to my credentialsFile>")

    sparkConf
  }
}

【问题讨论】：

标签： apache-spark google-cloud-platform apache-spark-sql google-bigquery

【解决方案1】：

读取常规表格不需要bigquery.tables.create 权限。但是，您提供的代码示例暗示该表实际上是一个 BigQuery 视图。 BigQuery 视图是逻辑引用，它们不会在服务器端具体化，为了让 spark 读取它们，它们首先需要具体化到临时表中。为了创建这个临时表，需要bigquery.tables.create 权限。

【讨论】：

嗨，我只受困于此。但是告诉我一件事 - 对于 bq 查询来说不是这样，因为使用 bq cmd 我能够从具有相同用户角色和凭据的视图中获取数据？但 PySpark 并非如此。当我尝试在 pyspark df 中加载和显示数据时，它给了我上述权限错误。
使用 bq 命令时，您使用的是 BigQuery API 并且需要 bigqeury.tables.* 权限。在 Spark 中，您使用的是 BigQuery Storage API 并且需要 bigquery.readsessions.* 权限。我会从那开始。有关更多详细信息，请打开一个带有错误消息的新问题

【解决方案2】：

检查下面的代码。

凭据

val credentials = """
     | {
     |           "type": "service_account",
     |           "project_id": "your project id",
     |           "private_key_id": "your private_key_id",
     |           "private_key": "-----BEGIN PRIVATE KEY-----\nxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx\n-----END PRIVATE KEY-----\n",
     |           "client_email": "xxxxx@company.com",
     |           "client_id": "111111111111111111111111111",
     |           "auth_uri": "https://accounts.google.com/o/oauth2/auth",
     |           "token_uri": "https://oauth2.googleapis.com/token",
     |           "auth_provider_x509_cert_url": "https://www.googleapis.com/oauth2/v1/certs",
     |           "client_x509_cert_url": "https://www.googleapis.com/robot/v1/metadata/x509/xxxxx40vvvvvv.iam.gserviceaccount.com"
     |         }
     | """

编码 base64 并将其传递给 spark conf。

def base64(data: String) = {
    import java.nio.charset.StandardCharsets
    import java.util.Base64
    Base64.getEncoder.encodeToString(data.getBytes(StandardCharsets.UTF_8))
}

spark.conf.set("credentials",base64(credentials))

spark
      .read
      .options("parentProject","parentProject")
      .option("table","dataset.table")
      .format("bigquery")
      .load()

【讨论】：