0

我在 bluemix 使用 Apache-Spark 服务。我目前无法通过 spark-submit 访问我的对象存储。

我知道该文件存在并且可以通过 jupyter notebook 上的 swift 访问。我使用以下方法进行验证：

file_name = "swift://notebooks.spark/small.verbatim"
text_file = sc.textFile(file_name)
print "number of verbatims", text_file.count()

输出是：

number of verbatims 100

但是当我尝试对 spark-submit 做同样的事情时，我得到了一个错误。

这是我通过 spark-submit 提交的代码：

import sys, traceback
from pymongo import MongoClient
import time
from datetime import datetime
from pyspark import SparkContext
sc = SparkContext('local', 'Schedule Insight Extractor')

try:
    file_name = "swift://notebooks.spark/small.verbatim"  # small dataset
    text_file = sc.textFile(file_name)
    r = None
    r = "number of verbatims", text_file.count()
except:
    e = sys.exc_info()[0]
    print ("ERROR %s", e)
    traceback.print_exc(file=sys.stdout)

这会引发以下异常：

Traceback (most recent call last):
  File "/gpfs/fs01/user/sf6d-7c3a9c08343577-05540e1c503a/data/workdir/spark-driver-cece5080-17dd-48e4-9036-52788e5a7b77/test_spark_submit.py", line 20, in <module>
    r = "number of verbatims", text_file.count()
  File "/usr/local/src/spark160master/spark-1.6.0-bin-2.6.0/python/lib/pyspark.zip/pyspark/rdd.py", line 1004, in count
    return self.mapPartitions(lambda i: [sum(1 for _ in i)]).sum()
  File "/usr/local/src/spark160master/spark-1.6.0-bin-2.6.0/python/lib/pyspark.zip/pyspark/rdd.py", line 995, in sum
    return self.mapPartitions(lambda x: [sum(x)]).fold(0, operator.add)
  File "/usr/local/src/spark160master/spark-1.6.0-bin-2.6.0/python/lib/pyspark.zip/pyspark/rdd.py", line 869, in fold
    vals = self.mapPartitions(func).collect()
  File "/usr/local/src/spark160master/spark-1.6.0-bin-2.6.0/python/lib/pyspark.zip/pyspark/rdd.py", line 771, in collect
    port = self.ctx._jvm.PythonRDD.collectAndServe(self._jrdd.rdd())
  File "/usr/local/src/spark160master/spark-1.6.0-bin-2.6.0/python/lib/py4j-0.9-src.zip/py4j/java_gateway.py", line 813, in __call__
    answer, self.gateway_client, self.target_id, self.name)
  File "/usr/local/src/spark160master/spark-1.6.0-bin-2.6.0/python/lib/py4j-0.9-src.zip/py4j/protocol.py", line 308, in get_return_value
    format(target_id, ".", name), value)
Py4JJavaError: An error occurred while calling z:org.apache.spark.api.python.PythonRDD.collectAndServe.
: java.lang.NullPointerException
        at org.apache.commons.httpclient.HttpMethodBase.getStatusCode(HttpMethodBase.java:570)
        at org.apache.hadoop.fs.swift.exceptions.SwiftInvalidResponseException.<init>(SwiftInvalidResponseException.java:53)
        at org.apache.hadoop.fs.swift.http.SwiftRestClient.buildException(SwiftRestClient.java:1827)
        at org.apache.hadoop.fs.swift.http.SwiftRestClient.perform(SwiftRestClient.java:1728)
        at org.apache.hadoop.fs.swift.http.SwiftRestClient.perform(SwiftRestClient.java:1662)
        at org.apache.hadoop.fs.swift.http.SwiftRestClient.authenticate(SwiftRestClient.java:1154)
        at org.apache.hadoop.fs.swift.http.SwiftRestClient.authIfNeeded(SwiftRestClient.java:1618)
        at org.apache.hadoop.fs.swift.http.SwiftRestClient.preRemoteCommand(SwiftRestClient.java:1634)
        at org.apache.hadoop.fs.swift.http.SwiftRestClient.headRequest(SwiftRestClient.java:1085)
        at org.apache.hadoop.fs.swift.snative.SwiftNativeFileSystemStore.stat(SwiftNativeFileSystemStore.java:258)
        at org.apache.hadoop.fs.swift.snative.SwiftNativeFileSystemStore.getObjectMetadata(SwiftNativeFileSystemStore.java:213)
        at org.apache.hadoop.fs.swift.snative.SwiftNativeFileSystemStore.getObjectMetadata(SwiftNativeFileSystemStore.java:182)
        at org.apache.hadoop.fs.swift.snative.SwiftNativeFileSystem.getFileStatus(SwiftNativeFileSystem.java:174)
        at org.apache.hadoop.fs.Globber.getFileStatus(Globber.java:57)
        at org.apache.hadoop.fs.Globber.glob(Globber.java:252)
        at org.apache.hadoop.fs.FileSystem.globStatus(FileSystem.java:1644)
        at org.apache.hadoop.mapred.FileInputFormat.singleThreadedListStatus(FileInputFormat.java:257)
        at org.apache.hadoop.mapred.FileInputFormat.listStatus(FileInputFormat.java:228)
        at org.apache.hadoop.mapred.FileInputFormat.getSplits(FileInputFormat.java:313)
        at org.apache.spark.rdd.HadoopRDD.getPartitions(HadoopRDD.scala:199)
        at org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:239)
        at org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:237)
        at scala.Option.getOrElse(Option.scala:120)
        at org.apache.spark.rdd.RDD.partitions(RDD.scala:237)
        at org.apache.spark.rdd.MapPartitionsRDD.getPartitions(MapPartitionsRDD.scala:35)
        at org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:239)
        at org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:237)
        at scala.Option.getOrElse(Option.scala:120)
        at org.apache.spark.rdd.RDD.partitions(RDD.scala:237)
        at org.apache.spark.api.python.PythonRDD.getPartitions(PythonRDD.scala:58)
        at org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:239)
        at org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:237)
        at scala.Option.getOrElse(Option.scala:120)
        at org.apache.spark.rdd.RDD.partitions(RDD.scala:237)
        at org.apache.spark.SparkContext.runJob(SparkContext.scala:1934)
        at org.apache.spark.rdd.RDD$$anonfun$collect$1.apply(RDD.scala:927)
        at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:150)
        at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:111)
        at org.apache.spark.rdd.RDD.withScope(RDD.scala:316)
        at org.apache.spark.rdd.RDD.collect(RDD.scala:926)
        at org.apache.spark.api.python.PythonRDD$.collectAndServe(PythonRDD.scala:405)
        at org.apache.spark.api.python.PythonRDD.collectAndServe(PythonRDD.scala)
        at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
        at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:95)
        at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:55)
        at java.lang.reflect.Method.invoke(Method.java:507)
        at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:231)
        at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:381)
        at py4j.Gateway.invoke(Gateway.java:259)
        at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:133)
        at py4j.commands.CallCommand.execute(CallCommand.java:79)
        at py4j.GatewayConnection.run(GatewayConnection.java:209)
        at java.lang.Thread.run(Thread.java:785)

我究竟做错了什么 ？我应该使用 SwiftClient 吗？网址结构不好吗？

4

2 回答 2

2

您正在尝试使用可在 Bluemix 服务上的 Notebook Enviornment 中使用的默认 hadoop 配置“spark”访问 notebooks 容器。

使用 spark-submit，您实际上是在调用一个新的 spark 上下文来使用。您需要明确设置新的对象存储凭据：-

def set_hadoop_config(creds):

prefix = "fs.swift.service." + creds['name']
hconf = sc._jsc.hadoopConfiguration()
hconf.set(prefix + ".auth.url", creds['auth_url'] + '/v2.0/tokens')
hconf.set(prefix + ".auth.endpoint.prefix", "endpoints")
hconf.set(prefix + ".tenant", creds['project_id'])
hconf.set(prefix + ".username", creds['user_id'])
hconf.set(prefix + ".password", creds['password'])
hconf.setInt(prefix + ".http.port", 8080)
hconf.set(prefix + ".region", creds['region'])
hconf.setBoolean(prefix + ".public", True)

ObjectSCredentials = {

'username': 'XXXXXXXXXXX',
'password': 'XXXXXXXXXXX',
'auth_url': 'https://identity.open.softlayer.com',
'project': 'XXXXXXXXXXX',
'project_id': 'XXXXXXXXXXX',
'region': 'dallas',
'user_id': 'XXXXXXXXXXX',
'domain_id': 'XXXXXXXXXXX',
'domain_name': 'XXXXXXXXXXX',
'filename': 'small.verbatim',
'container': 'notebooks',
'tenantId': ‘XXXXXXXXX'

}

ObjectSCredentials['name'] = ’TEST'

set_hadoop_config(ObjectSCredentials)

rdddata = sc.textFile("swift://notebooks." + ObjectSCredentials['name'] + “/small.verbatim")

谢谢，

查尔斯。

于 2016-04-05T16:32:21.153 回答
1

您可以将以下内容用于 auth url 属性：替换

hconf.set(prefix + ".auth.url", creds['auth_url'] + '/v2.0/tokens')

和

hconf.set(prefix + ".auth.url", creds['auth_url']+'/v3/auth/tokens')
于 2016-05-19T20:45:26.177 回答