如何在本地使用 Spark BigQuery 连接器？

Question

出于测试目的，我想使用BigQuery Connector 在 BigQuery 中编写 Parquet Avro 日志。在我写的时候，没有办法直接从 UI 读取 Parquet 来摄取它，所以我正在编写一个 Spark 作业来这样做。

在 Scala 中，工作主体暂时如下：

val events: RDD[RichTrackEvent] =
readParquetRDD[RichTrackEvent, RichTrackEvent](sc, googleCloudStorageUrl)

val conf = sc.hadoopConfiguration
conf.set("mapred.bq.project.id", "myproject")

// Output parameters
val projectId = conf.get("fs.gs.project.id")
val outputDatasetId = "logs"
val outputTableId = "test"
val outputTableSchema = LogSchema.schema

// Output configuration
BigQueryConfiguration.configureBigQueryOutput(
  conf, projectId, outputDatasetId, outputTableId, outputTableSchema
)
conf.set(
  "mapreduce.job.outputformat.class",
  classOf[BigQueryOutputFormat[_, _]].getName
)

events
  .mapPartitions {
    items =>
      val gson = new Gson()
      items.map(e => gson.fromJson(e.toString, classOf[JsonObject]))
  }
  .map(x => (null, x))
  .saveAsNewAPIHadoopDataset(conf)

由于 BigQueryOutputFormat 没有找到 Google 凭据，它会回退到元数据主机上，尝试使用以下堆栈跟踪来发现它们：

016-06-13 11:40:53 WARN  HttpTransport:993 - exception thrown while executing request
java.net.UnknownHostException: metadata
at java.net.AbstractPlainSocketImpl.connect(AbstractPlainSocketImpl.java:184)
at java.net.SocksSocketImpl.connect(SocksSocketImpl.java:392)
at java.net.Socket.connect(Socket.java:589    at sun.net.NetworkClient.doConnect(NetworkClient.java:175)
at sun.net.www.http.HttpClient.openServer(HttpClient.java:432)
at sun.net.www.http.HttpClient.openServer(HttpClient.java:527)
at sun.net.www.http.HttpClient.<init>(HttpClient.java:211)
at sun.net.www.http.HttpClient.New(HttpClient.java:308)
at sun.net.www.http.HttpClient.New(HttpClient.java:326)
at sun.net.www.protocol.http.HttpURLConnection.getNewHttpClient(HttpURLConnection.java:1169)
at sun.net.www.protocol.http.HttpURLConnection.plainConnect0(HttpURLConnection.java:1105)
at sun.net.www.protocol.http.HttpURLConnection.plainConnect(HttpURLConnection.java:999)
at sun.net.www.protocol.http.HttpURLConnection.connect(HttpURLConnection.java:933)
at com.google.api.client.http.javanet.NetHttpRequest.execute(NetHttpRequest.java:93)
at com.google.api.client.http.HttpRequest.execute(HttpRequest.java:972)
at com.google.cloud.hadoop.util.CredentialFactory$ComputeCredentialWithRetry.executeRefreshToken(CredentialFactory.java:160)
at com.google.api.client.auth.oauth2.Credential.refreshToken(Credential.java:489)
at com.google.cloud.hadoop.util.CredentialFactory.getCredentialFromMetadataServiceAccount(CredentialFactory.java:207)
at com.google.cloud.hadoop.util.CredentialConfiguration.getCredential(CredentialConfiguration.java:72)
at com.google.cloud.hadoop.io.bigquery.BigQueryFactory.createBigQueryCredential(BigQueryFactory.java:81)
at com.google.cloud.hadoop.io.bigquery.BigQueryFactory.getBigQuery(BigQueryFactory.java:101)
at com.google.cloud.hadoop.io.bigquery.BigQueryFactory.getBigQueryHelper(BigQueryFactory.java:89)
at com.google.cloud.hadoop.io.bigquery.BigQueryOutputCommitter.<init>(BigQueryOutputCommitter.java:70)
at com.google.cloud.hadoop.io.bigquery.BigQueryOutputFormat.getOutputCommitter(BigQueryOutputFormat.java:102)
at com.google.cloud.hadoop.io.bigquery.BigQueryOutputFormat.getOutputCommitter(BigQueryOutputFormat.java:84)
at com.google.cloud.hadoop.io.bigquery.BigQueryOutputFormat.getOutputCommitter(BigQueryOutputFormat.java:30)
at org.apache.spark.rdd.PairRDDFunctions$$anonfun$saveAsNewAPIHadoopDataset$1.apply$mcV$sp(PairRDDFunctions.scala:1135)
at org.apache.spark.rdd.PairRDDFunctions$$anonfun$saveAsNewAPIHadoopDataset$1.apply(PairRDDFunctions.scala:1078)
at org.apache.spark.rdd.PairRDDFunctions$$anonfun$saveAsNewAPIHadoopDataset$1.apply(PairRDDFunctions.scala:1078)
at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:112)
at org.apache.spark.rdd.RDD.withScope(RDD.scala:357)
at org.apache.spark.rdd.PairRDDFunctions.saveAsNewAPIHadoopDataset(PairRDDFunctions.scala:1078)

这当然是意料之中的，但它应该能够使用我的服务帐户及其密钥，因为 GoogleCredential.getApplicationDefault() 返回从 GOOGLE_APPLICATION_CREDENTIALS 环境变量获取的适当凭据。

由于连接器似乎从 hadoop 配置中读取凭据，因此要设置什么键才能读取 GOOGLE_APPLICATION_CREDENTIALS ？有没有办法配置输出格式以使用提供的 GoogleCredential 对象？

Answer 1

如果我正确理解您的问题 - 您可能需要设置：

<name>mapred.bq.auth.service.account.enable</name>
<name>mapred.bq.auth.service.account.email</name>
<name>mapred.bq.auth.service.account.keyfile</name>
<name>mapred.bq.project.id</name>
<name>mapred.bq.gcs.bucket</name>

这里，mapred.bq.auth.service.account.keyfile 应该指向旧式“P12”密钥文件的完整文件路径；或者，如果您使用较新的“JSON”密钥文件，则应将“电子邮件”和“密钥文件”条目替换为单个 mapred.bq.auth.service.account.json.keyfile 密钥：

<name>mapred.bq.auth.service.account.enable</name>
<name>mapred.bq.auth.service.account.json.keyfile</name>
<name>mapred.bq.project.id</name>
<name>mapred.bq.gcs.bucket</name>

您可能还想看看https://github.com/spotify/spark-bigquery - 这是使用 BQ 和 Spark 的更文明的方式。在这种情况下使用的 setGcpJsonKeyFile 方法与您为 mapred.bq.auth.service.account.json.keyfile 设置的 JSON 文件相同（如果使用 Hadoop 的 BQ 连接器）。