如何通过Spark SQL连接BigQuery？

由于需要使用SQLContext对象来使用Spark SQL，因此需要首先配置SparkContext以连接到BigQuery。从我的角度来看，BigQuery Connector（由sramalingam24和Kenneth Jung解决）可用于在BigQuery中查询数据。

请注意，sramalingam24提供了一个示例链接，以下是代码摘要：

bucket = sc._jsc.hadoopConfiguration().get('fs.gs.system.bucket')
project = sc._jsc.hadoopConfiguration().get('fs.gs.project.id')
input_directory = 'gs://{}/hadoop/tmp/bigquery/pyspark_input'.format(bucket)

conf = {

# Input Parameters.
'mapred.bq.project.id': project,
'mapred.bq.gcs.bucket': bucket,
'mapred.bq.temp.gcs.path': input_directory,
'mapred.bq.input.project.id': 'publicdata',
'mapred.bq.input.dataset.id': 'samples',
'mapred.bq.input.table.id': 'shakespeare',

}

table_data = sc.newAPIHadoopRDD(

'com.google.cloud.hadoop.io.bigquery.JsonTextBigQueryInputFormat',
'org.apache.hadoop.io.LongWritable',
'com.google.gson.JsonObject',
conf=conf)

word_counts = (

table_data
.map(lambda record: json.loads(record[1]))
.map(lambda x: (x['word'].lower(), int(x['word_count'])))
.reduceByKey(lambda x, y: x + y))

sql_context = SQLContext(sc)
(word_counts
.toDF(['word', 'word_count'])
.write.format('json').save(output_directory))
然后，您可以下载其他Hadoop集群的连接器jar。Kenneth Jung提供的链接信息表明选项--jar可用于包含连接符（--jars gs：//spark-lib/bigquery/spark-bigquery-latest.jar），这是一个包含的选项驱动程序和执行程序类路径上的jar。