在 DataLab 笔记本中使用 Dataproc 和 Spark BigQuery 连接器时出错
Error using Dataproc and Spark BigQuery Connector within DataLab Notebooks
我能够在 Google Chrome 中获得 Google Datalab（笔记本）运行 并具有正确的 TCP 防火墙权限。使用简单的脚本，这将启动最新的 spark 集群（1 个主节点和 3 个使用 Dataproc 的工作节点）。首先，我们在 spark-submit 中测试以下代码，然后在启动 DataLab 后，我不确定如何修复以下错误。
第一步：从云端启动 Dataproc 集群 Shell
gcloud dataproc clusters create cluster1021 \
--subnet default --zone us-west1-a \
--master-machine-type n1-standard-2 \
--master-boot-disk-size 30 --num-workers 2 \
--worker-machine-type n1-standard-2 \
--worker-boot-disk-size 30 --image-version 1.3-deb9 \
--project bigdata-228217 \
--initialization-actions 'gs://dataproc-initialization-actions/datalab/datalab.sh','gs://dataproc-initialization-actions/connectors/connectors.sh' \
--metadata 'gcs-connector-version=1.9.11' \
--metadata 'bigquery-connector-version=0.13.11'
成功启动后，我进行了测试，发现 Bigquery 连接器正在使用来自 Google here.
的 spark-submit wordcount.py 示例
第二步：将此代码作为 wordcount.py 和 touch wordcount.py 包含在主目录中，然后从 nano wordcount.py 粘贴下面的代码并保存。
#!/usr/bin/python
"""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()
# Use the Cloud Storage bucket for temporary BigQuery export data used
# by the InputFormat. This assumes the Cloud Storage connector for
# Hadoop is configured.
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',
}
# Output Parameters.
output_dataset = 'wordcount_dataset'
output_table = 'wordcount_output'
# Load data in from BigQuery.
table_data = sc.newAPIHadoopRDD(
'com.google.cloud.hadoop.io.bigquery.JsonTextBigQueryInputFormat',
'org.apache.hadoop.io.LongWritable',
'com.google.gson.JsonObject',
conf=conf)
# Perform word count.
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))
# Display 10 results.
pprint.pprint(word_counts.take(10))
# Stage data formatted as newline-delimited JSON in Cloud Storage.
output_directory = 'gs://{}/hadoop/tmp/bigquery/pyspark_output'.format(bucket)
output_files = output_directory + '/part-*'
sql_context = SQLContext(sc)
(word_counts
.toDF(['word', 'word_count'])
.write.format('json').save(output_directory))
# Shell out to bq CLI to perform BigQuery import.
subprocess.check_call(
'bq load --source_format NEWLINE_DELIMITED_JSON '
'--replace '
'--autodetect '
'{dataset}.{table} {files}'.format(
dataset=output_dataset, table=output_table, files=output_files
).split())
# Manually clean up the staging_directories, otherwise BigQuery
# files will remain indefinitely.
input_path = sc._jvm.org.apache.hadoop.fs.Path(input_directory)
input_path.getFileSystem(sc._jsc.hadoopConfiguration()).delete(input_path, True)
output_path = sc._jvm.org.apache.hadoop.fs.Path(output_directory)
output_path.getFileSystem(sc._jsc.hadoopConfiguration()).delete(
output_path, True)
现在，shell，spark-submit 的输出是结果 -- 显示 BigQuery 连接器工作。
spark-submit wordcount.py
...
(pinnace,3)
(bone,21)
(lug,2)
(vailing,2)
(bombast,3)
(gaping,11)
(hem,5)
('non,1)
(stinks,1)
(forsooth,48)
步骤 3 设置防火墙以允许在浏览器中查看 TCP DataLab
为 DataLab 创建防火墙规则
在设置页面上，您将为 DataLab 防火墙规则创建一个名称，并允许使用以下 TCP 端口，并在您的网络 IP 地址后紧跟“/32” - 您可以找到 here。
第 4 步：使用 <YOUR IP>:8080 在 Google Chrome 中启动 DataLab，您应该会看到 DataLab Notebook
你会看到这个。现在打开一个新笔记本，在第一个单元格中我停止了 spark 上下文并将上面的莎士比亚代码粘贴到第二个单元格中。
这是输出。问题，我需要做什么才能让 Bigquery 连接器在 Datalab 中与 Pyspark 一起工作？
Py4JJavaErrorTraceback (most recent call last)
<ipython-input-4-62761a09a7c5> in <module>()
36 'org.apache.hadoop.io.LongWritable',
37 'com.google.gson.JsonObject',
---> 38 conf=conf)
39
40 # Perform word count.
/usr/lib/spark/python/lib/pyspark.zip/pyspark/context.py in newAPIHadoopRDD(self, inputFormatClass, keyClass, valueClass, keyConverter, valueConverter, conf, batchSize)
735 jrdd = self._jvm.PythonRDD.newAPIHadoopRDD(self._jsc, inputFormatClass, keyClass,
736 valueClass, keyConverter, valueConverter,
--> 737 jconf, batchSize)
738 return RDD(jrdd, self)
739
/usr/lib/spark/python/lib/py4j-0.10.7-src.zip/py4j/java_gateway.py in __call__(self, *args)
1255 answer = self.gateway_client.send_command(command)
1256 return_value = get_return_value(
-> 1257 answer, self.gateway_client, self.target_id, self.name)
1258
1259 for temp_arg in temp_args:
/usr/lib/spark/python/lib/pyspark.zip/pyspark/sql/utils.py in deco(*a, **kw)
61 def deco(*a, **kw):
62 try:
---> 63 return f(*a, **kw)
64 except py4j.protocol.Py4JJavaError as e:
65 s = e.java_exception.toString()
/usr/lib/spark/python/lib/py4j-0.10.7-src.zip/py4j/protocol.py in get_return_value(answer, gateway_client, target_id, name)
326 raise Py4JJavaError(
327 "An error occurred while calling {0}{1}{2}.\n".
--> 328 format(target_id, ".", name), value)
329 else:
330 raise Py4JError(
Py4JJavaError: An error occurred while calling z:org.apache.spark.api.python.PythonRDD.newAPIHadoopRDD.
: java.lang.ClassNotFoundException: com.google.cloud.hadoop.io.bigquery.JsonTextBigQueryInputFormat
at java.net.URLClassLoader.findClass(URLClassLoader.java:382)
at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
at java.lang.Class.forName0(Native Method)
at java.lang.Class.forName(Class.java:348)
at org.apache.spark.util.Utils$.classForName(Utils.scala:239)
at org.apache.spark.api.python.PythonRDD$.newAPIHadoopRDDFromClassNames(PythonRDD.scala:313)
at org.apache.spark.api.python.PythonRDD$.newAPIHadoopRDD(PythonRDD.scala:296)
at org.apache.spark.api.python.PythonRDD.newAPIHadoopRDD(PythonRDD.scala)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)
at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)
at py4j.Gateway.invoke(Gateway.java:282)
at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
at py4j.commands.CallCommand.execute(CallCommand.java:79)
at py4j.GatewayConnection.run(GatewayConnection.java:238)
at java.lang.Thread.run(Thread.java:748)
根据 this line 判断，Datalab init 操作将 BQ 和 GCS 连接器安装到 Docker 容器中。
因为 Dataproc 1.3 默认不附带 BQ 连接器，并且因为您指定了连接器初始化操作，在集群上安装 BQ 连接器，在 DataLab 初始化操作后，Docker 无法将 BQ 连接器安装到 Datalab 容器中在 Datalab 初始化操作执行期间。
要解决此问题，您需要更改初始化操作的顺序：
gcloud dataproc clusters create \
. . .
--initialization-actions=gs://dataproc-initialization-actions/datalab/connectors.sh,gs://dataproc-initialization-actions/connectors/datalab.sh
作为一项小改进，您目前不需要指定 GCS 连接器版本 (--metadata 'gcs-connector-version=1.9.11')，因为最新的 Dataproc 1.3 映像已经预装了 GCS 连接器 1.9.11。
我能够在 Google Chrome 中获得 Google Datalab（笔记本）运行 并具有正确的 TCP 防火墙权限。使用简单的脚本，这将启动最新的 spark 集群（1 个主节点和 3 个使用 Dataproc 的工作节点）。首先，我们在 spark-submit 中测试以下代码，然后在启动 DataLab 后，我不确定如何修复以下错误。
第一步：从云端启动 Dataproc 集群 Shell
gcloud dataproc clusters create cluster1021 \
--subnet default --zone us-west1-a \
--master-machine-type n1-standard-2 \
--master-boot-disk-size 30 --num-workers 2 \
--worker-machine-type n1-standard-2 \
--worker-boot-disk-size 30 --image-version 1.3-deb9 \
--project bigdata-228217 \
--initialization-actions 'gs://dataproc-initialization-actions/datalab/datalab.sh','gs://dataproc-initialization-actions/connectors/connectors.sh' \
--metadata 'gcs-connector-version=1.9.11' \
--metadata 'bigquery-connector-version=0.13.11'
成功启动后，我进行了测试，发现 Bigquery 连接器正在使用来自 Google here.
的spark-submit wordcount.py 示例
第二步：将此代码作为 wordcount.py 和 touch wordcount.py 包含在主目录中，然后从 nano wordcount.py 粘贴下面的代码并保存。
#!/usr/bin/python
"""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()
# Use the Cloud Storage bucket for temporary BigQuery export data used
# by the InputFormat. This assumes the Cloud Storage connector for
# Hadoop is configured.
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',
}
# Output Parameters.
output_dataset = 'wordcount_dataset'
output_table = 'wordcount_output'
# Load data in from BigQuery.
table_data = sc.newAPIHadoopRDD(
'com.google.cloud.hadoop.io.bigquery.JsonTextBigQueryInputFormat',
'org.apache.hadoop.io.LongWritable',
'com.google.gson.JsonObject',
conf=conf)
# Perform word count.
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))
# Display 10 results.
pprint.pprint(word_counts.take(10))
# Stage data formatted as newline-delimited JSON in Cloud Storage.
output_directory = 'gs://{}/hadoop/tmp/bigquery/pyspark_output'.format(bucket)
output_files = output_directory + '/part-*'
sql_context = SQLContext(sc)
(word_counts
.toDF(['word', 'word_count'])
.write.format('json').save(output_directory))
# Shell out to bq CLI to perform BigQuery import.
subprocess.check_call(
'bq load --source_format NEWLINE_DELIMITED_JSON '
'--replace '
'--autodetect '
'{dataset}.{table} {files}'.format(
dataset=output_dataset, table=output_table, files=output_files
).split())
# Manually clean up the staging_directories, otherwise BigQuery
# files will remain indefinitely.
input_path = sc._jvm.org.apache.hadoop.fs.Path(input_directory)
input_path.getFileSystem(sc._jsc.hadoopConfiguration()).delete(input_path, True)
output_path = sc._jvm.org.apache.hadoop.fs.Path(output_directory)
output_path.getFileSystem(sc._jsc.hadoopConfiguration()).delete(
output_path, True)
现在，shell，spark-submit 的输出是结果 -- 显示 BigQuery 连接器工作。
spark-submit wordcount.py
...
(pinnace,3)
(bone,21)
(lug,2)
(vailing,2)
(bombast,3)
(gaping,11)
(hem,5)
('non,1)
(stinks,1)
(forsooth,48)
步骤 3 设置防火墙以允许在浏览器中查看 TCP DataLab
为 DataLab 创建防火墙规则
在设置页面上，您将为 DataLab 防火墙规则创建一个名称，并允许使用以下 TCP 端口，并在您的网络 IP 地址后紧跟“/32” - 您可以找到 here。
第 4 步：使用 <YOUR IP>:8080 在 Google Chrome 中启动 DataLab，您应该会看到 DataLab Notebook
你会看到这个。现在打开一个新笔记本，在第一个单元格中我停止了 spark 上下文并将上面的莎士比亚代码粘贴到第二个单元格中。
这是输出。问题，我需要做什么才能让 Bigquery 连接器在 Datalab 中与 Pyspark 一起工作？
Py4JJavaErrorTraceback (most recent call last)
<ipython-input-4-62761a09a7c5> in <module>()
36 'org.apache.hadoop.io.LongWritable',
37 'com.google.gson.JsonObject',
---> 38 conf=conf)
39
40 # Perform word count.
/usr/lib/spark/python/lib/pyspark.zip/pyspark/context.py in newAPIHadoopRDD(self, inputFormatClass, keyClass, valueClass, keyConverter, valueConverter, conf, batchSize)
735 jrdd = self._jvm.PythonRDD.newAPIHadoopRDD(self._jsc, inputFormatClass, keyClass,
736 valueClass, keyConverter, valueConverter,
--> 737 jconf, batchSize)
738 return RDD(jrdd, self)
739
/usr/lib/spark/python/lib/py4j-0.10.7-src.zip/py4j/java_gateway.py in __call__(self, *args)
1255 answer = self.gateway_client.send_command(command)
1256 return_value = get_return_value(
-> 1257 answer, self.gateway_client, self.target_id, self.name)
1258
1259 for temp_arg in temp_args:
/usr/lib/spark/python/lib/pyspark.zip/pyspark/sql/utils.py in deco(*a, **kw)
61 def deco(*a, **kw):
62 try:
---> 63 return f(*a, **kw)
64 except py4j.protocol.Py4JJavaError as e:
65 s = e.java_exception.toString()
/usr/lib/spark/python/lib/py4j-0.10.7-src.zip/py4j/protocol.py in get_return_value(answer, gateway_client, target_id, name)
326 raise Py4JJavaError(
327 "An error occurred while calling {0}{1}{2}.\n".
--> 328 format(target_id, ".", name), value)
329 else:
330 raise Py4JError(
Py4JJavaError: An error occurred while calling z:org.apache.spark.api.python.PythonRDD.newAPIHadoopRDD.
: java.lang.ClassNotFoundException: com.google.cloud.hadoop.io.bigquery.JsonTextBigQueryInputFormat
at java.net.URLClassLoader.findClass(URLClassLoader.java:382)
at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
at java.lang.Class.forName0(Native Method)
at java.lang.Class.forName(Class.java:348)
at org.apache.spark.util.Utils$.classForName(Utils.scala:239)
at org.apache.spark.api.python.PythonRDD$.newAPIHadoopRDDFromClassNames(PythonRDD.scala:313)
at org.apache.spark.api.python.PythonRDD$.newAPIHadoopRDD(PythonRDD.scala:296)
at org.apache.spark.api.python.PythonRDD.newAPIHadoopRDD(PythonRDD.scala)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)
at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)
at py4j.Gateway.invoke(Gateway.java:282)
at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
at py4j.commands.CallCommand.execute(CallCommand.java:79)
at py4j.GatewayConnection.run(GatewayConnection.java:238)
at java.lang.Thread.run(Thread.java:748)
根据 this line 判断，Datalab init 操作将 BQ 和 GCS 连接器安装到 Docker 容器中。
因为 Dataproc 1.3 默认不附带 BQ 连接器，并且因为您指定了连接器初始化操作，在集群上安装 BQ 连接器，在 DataLab 初始化操作后，Docker 无法将 BQ 连接器安装到 Datalab 容器中在 Datalab 初始化操作执行期间。
要解决此问题，您需要更改初始化操作的顺序：
gcloud dataproc clusters create \
. . .
--initialization-actions=gs://dataproc-initialization-actions/datalab/connectors.sh,gs://dataproc-initialization-actions/connectors/datalab.sh
作为一项小改进，您目前不需要指定 GCS 连接器版本 (--metadata 'gcs-connector-version=1.9.11')，因为最新的 Dataproc 1.3 映像已经预装了 GCS 连接器 1.9.11。