我正在尝试使用Dataproc在PySpark中运行脚本。
该脚本是此示例与我需要执行的操作之间的合并,因为我想检查是否一切正常。显然,事实并非如此。
我得到的错误是:
我文件 "/usr/lib/spark/python/lib/py4j-0.10.7-src.zip/py4j/protocol.py",第 328 行,get_return_value py4j.protocol.Py4JJavaError:调用 z:org.apache.spark.api.python.PythonRDD.newAPIHadoopRDD 时出错。 : java.lang.ClassNotFoundException: com.google.cloud.hadoop.io.bigquery.JsonTextBigQueryInputFormat
确保我拥有所有的罐子,并按照其他类似帖子中的建议添加了一些新罐子。我还检查了SPARK_HOME
变量。
下面可以看到代码;尝试实例化table_data时出现错误。
"""BigQuery I/O PySpark example."""
from __future__ import absolute_import
import json
import pprint
import subprocess
import pyspark
from pyspark.sql import SQLContext
sc = pyspark.SparkContext()
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 = {
'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',
}
output_dataset = 'wordcount_dataset'
output_table = 'wordcount_output'
table_data = sc.newAPIHadoopRDD(
'com.google.cloud.hadoop.io.bigquery.JsonTextBigQueryInputFormat',
'org.apache.hadoop.io.LongWritable',
'com.google.gson.JsonObject',
conf=conf)
如示例中所述,提交作业时需要包含 BigQuery 连接器 jar。
通过 Dataproc 作业 API:
gcloud dataproc jobs submit pyspark --cluster=${CLUSTER}
/path/to/your/script.py
--jars=gs://hadoop-lib/bigquery/bigquery-connector-hadoop2-latest.jar
或从群集内部spark-submit
:
spark-submit --jars=gs://hadoop-lib/bigquery/bigquery-connector-hadoop2-latest.jar
/path/to/your/script.py