apache-spark - 通过 Hadoop 输入格式示例用于 pyspark 的 BigQuery 连接器

标签 apache-spark google-bigquery pyspark google-hadoop google-cloud-dataproc

我有一个存储在 BigQuery 表中的大型数据集,我想将其加载到 pypark RDD 中以进行 ETL 数据处理。

我意识到 BigQuery 支持 Hadoop 输入/输出格式

https://cloud.google.com/hadoop/writing-with-bigquery-connector

并且 pyspark 应该能够使用此接口(interface),以便通过使用“newAPIHadoopRDD”方法创建 RDD。

http://spark.apache.org/docs/latest/api/python/pyspark.html

不幸的是,两端的文档似乎很少,超出了我对 Hadoop/Spark/BigQuery 的了解。有没有人知道如何做到这一点?

最佳答案

Google 现在有一个 example关于如何将 BigQuery 连接器与 Spark 结合使用。

使用 GsonBigQueryInputFormat 似乎确实存在问题,但我得到了一个简单的莎士比亚字数统计示例

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)

关于apache-spark - 通过 Hadoop 输入格式示例用于 pyspark 的 BigQuery 连接器,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/31401239/

相关文章:

python - PySpark - sortByKey() 方法以原始顺序从 k,v 对返回值

hadoop - BigQuery Hadoop 连接器和 Dataproc

apache-spark - 带有时区偏移的时间戳

python - PySpark 与 sklearn TFIDF

java - 如何使用RDD持久化和缓存?

java - 如何使用 MapReduce 函数通过 Spark 在 Java 中排序

google-bigquery - 如何通过 HTTP API 获取 BigQuery 查询的元数据

google-bigquery - 如何在 Google Bigquery 中将时间戳转换为日期数据类型

amazon-s3 - Spark的S3角色授权?

scala - Spark 中的广播变量