使用 pyspark 脚本从 bigquery 加载表到 spark 集群答案

【问题标题】：load table from bigquery to spark cluster with pyspark script使用 pyspark 脚本从 bigquery 加载表到 spark 集群
【发布时间】：2016-01-26 09:23:39
【问题描述】：

我在 bigquery 中加载了一个数据表，我想通过 pyspark .py 文件将其导入我的 Spark 集群。

我在Dataproc + BigQuery examples - any available? 中看到有一种方法可以使用 scala 在 spark 集群中加载 bigquery 表，但是有没有办法在 pyspark 脚本中执行此操作？

【问题讨论】：

标签： python apache-spark google-bigquery pyspark google-cloud-dataproc

【解决方案1】：

这来自@MattJ 在this question。这是连接到 Spark 中的 BigQuery 并执行字数统计的示例。

import json
import pyspark
sc = pyspark.SparkContext()

hadoopConf=sc._jsc.hadoopConfiguration()
hadoopConf.get("fs.gs.system.bucket")

conf = {"mapred.bq.project.id": "<project_id>", "mapred.bq.gcs.bucket": "<bucket>",
    "mapred.bq.input.project.id": "publicdata", 
    "mapred.bq.input.dataset.id":"samples", 
    "mapred.bq.input.table.id": "shakespeare"  }

tableData = sc.newAPIHadoopRDD(
    "com.google.cloud.hadoop.io.bigquery.JsonTextBigQueryInputFormat",
    "org.apache.hadoop.io.LongWritable", "com.google.gson.JsonObject", 
    conf=conf).map(lambda k: json.loads(k[1])).map(lambda x: (x["word"],
    int(x["word_count"]))).reduceByKey(lambda x,y: x+y)

print tableData.take(10)

您需要更改<project_id> 和<bucket> 以匹配您的项目的设置。

【讨论】：