Spark BigQuery Connectorをローカルに使用するには？

Question

テスト目的のために、BigQuery Connectorを使用して、BigQueryのParquet Avroログを書きます。私が書いているように、UIから直接Parquetを読んでそれを摂取する方法はないので、そうするためにSparkの仕事を書いています。 Scalaで

は、当分の間、仕事のボディは以下の通りです：BigQueryOutputFormatは、Googleの資格情報を見つけていないので

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) 

、それは以下のスタックトレースでそれらを発見しようとするメタデータをホストにフォールバック：それは期待もちろんですが、GOOGLE_APPLICATION_CREDENTIALS環境変数からフェッチGoogleCredential.getApplicationDefault()戻って適切な資格としての私のサービスアカウントとそのキーを使用することができるはずです

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) 

。

コネクタが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メソッドは、HadoopにBQコネクタを使用する場合はmapred.bq.auth.service.account.json.keyfileに設定したのと同じJSONファイルです。