python – 使用pyspark脚本从bigquery加载表到spark集群

2019-06-09 04:46:38 阅读：360 来源： 互联网

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

我有一个在bigquery中加载的数据表,我想通过pyspark .py文件在我的spark集群中导入它.

我在Dataproc + BigQuery examples – any available?中看到有一种方法可以使用scala在spark集群中加载一个bigquery表,但有没有办法在pyspark脚本中执行它？

解决方法:

这来自@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>匹配项目的设置.

标签：python,apache-spark,pyspark,google-bigquery,google-cloud-dataproc
来源： https://codeday.me/bug/20190609/1202703.html

本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享；
2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关；
3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关；
4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除；
5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。

ICode9

python – 使用pyspark脚本从bigquery加载表到spark集群