| 各行各业都在以惊人的速度生成数据，包括测序系统生成的基因组数据、具有超高清格式的媒体和娱乐数据，以及众多传感器生成的物联网
(IoT) 数据。IBM Cloud Object Storage（IBM COS，以前称为 Cleversafe）技术为这些应用提供了高容量、经济有效的存储。但仅存储数据还不够；还需要从数据中获取价值；可以使用领先的数据分析处理引擎
Apache Spark 来实现此目的。Spark 的运行速度是 Hadoop MapReduce
的 100 倍，而且它还结合了 SQL、流处理和复杂情况分析。
本文将介绍如何让 Spark 能对 IBM COS 中存储的数据进行分析。我们将介绍如何使用 Stocator
和 OpenStack Keystone，前者是一个用作驱动程序的开源软件，后者提供了身份验证功能。Stocator
利用了对象存储语义，而且与以前专为处理文件系统而设计的 Spark 存储连接器相比，Stocator
显著提高了性能。Stocator 采用 JOSS（一个开源 Java 客户端）生成 HTTP REST
命令，这些命令通过 OpenStack Swift 接口访问 IBM COS。
下图演示了 IBM COS、Stocator 与 OpenStack Keystone 之间的三角关系。
安装和配置 Spark
下载 Spark。Spark 网站提供了构建、安装和配置 Spark 的操作说明。依据您的设置，可以将
Spark 配置为一台独立机器，或者在集群上使用 YARN、Mesos 或 Spark 的独立集群管理器。在我们的示例中，我们结合使用了
IBM COS 和 Spark 2.0.1。
安装和配置 IBM COS
安装 Cloud Object Storage (COS)。在我们的示例中，我们为 IBM COS
设置了 Keystone 身份验证。
安装和配置 Stocator
为了从 Spark 访问 IBM COS，我们使用了开源驱动程序软件 Stocator。Stocator
是 Spark 的高性能的对象存储连接器，它利用了对象存储语义。它提供了 OpenStack Swift
API 的完整驱动程序，可轻松地扩展它来支持其他对象存储接口。我们利用了 Stocator 通过其
Swift API 将 Spark 与 IBM COS 相连的能力。
要使用 Stocator，请完成以下步骤。
1.从 https://github.com/SparkTC/stocator
下载源代码，使用 git 复制或克隆它。
2.从 Stocator 的目录输入 mvn clean package
–Pall-in-one 来构建 Stocator。
要将 Spark 配置为利用 Stocator 访问 IBM COS，需要定义 Stocator 及其设置。有两种配置
Stocator 的方法：
1.向配置文件添加参数
2.向代码添加参数
向配置文件添加参数
要创建 core-site.xml 配置文件，可执行以下操作之一：
1.使用 Stocator/conf 目录中的 core-site.xml.template
文件 作为模板
2.使用 Keystone Version 2 配置文件
3.使用 Keystone Version 3 配置文件
使用 Stocator/conf 目录中的 core-site.xml.template
文件
在 Configuration Files 部分，通过输入以下命令，访问用于配置基于 Keystone
的身份验证的 core-site.xml 示例：
清单 1. 访问 core-site.xml.template 文件
| ofer@beginnings:~$
cd ~/stocator/conf
ofer@beginnings:~/stocator/conf$ cp core-site.xml.template
~/spark-2.0.1/conf/core-site.x |
使用 Keystone Version 2 配置文件
对于 Keystone Version 2，可以使用这个 core-site.xml 并按清单下方的说明进行替换。
清单 2. Keystone Version 2 配置文件
| <configuration>
<property>
<name>fs.swift2d.impl</name>
<value>com.ibm.stocator.fs.ObjectStoreFile
System</value>
</property>
<!-- Keystone based authentication -->
<property>
<name>fs.swift2d.service.spark.auth.url</name>
<value>http://your.keystone.server.com:
5000/v2.0/tokens</value>
</property>
<property>
<name>fs.swift2d.service.spark.public</name>
<value>true</value>
</property>
<property>
<name>fs.swift2d.service.spark.tenant</name>
<value>service</value>
</property>
<property>
<name>fs.swift2d.service.spark.password</name>
<value>passw0rd</value>
</property>
<property>
<name>fs.swift2d.service.spark.username</name>
<value>swift</value>
</property>
<property>
<name>fs.swift2d.service.spark.auth.method</name>
<value>keystone</value>
</property>
<property>
<name>fs.swift2d.service.spark.region</name>
<value>IBMCOS</value>
</property>
</configuration> |
1.将 your.keystone.server.com 替换为
Keystone 服务器的真实地址。
2.将所有身份验证凭证（tenant、username 和 password）替换为您的对象存储的有效凭证。
3.使用全球 Keystone 时，需要根据为 IBM COS
访问定义的 Keystone 地区来定义 region 属性。
使用 Keystone Version 3 配置文件
对于 Keystone Version 3，可以使用这个 core-site.xml 并按清单下方的说明进行替换。请记住，对于
Keystone Version 3，使用 userID 和 tenantID 代替 username
和 tenant 值。
清单 3. Keystone Version 3 配置文件
| <configuration>
<property>
<name>fs.swift2d.impl</name>
<value>com.ibm.stocator.fs.ObjectStoreFileSystem
</value>
</property>
<!-- Keystone based authentication -->
<property>
<name>fs.swift2d.service.spark.auth.url</name>
<value>http://your.keystone.server.com:5000/v3
/auth/tokens</value>
</property>
<property>
<name>fs.swift2d.service.spark.public</name>
<value>true</value>
</property>
<property>
<name>fs.swift2d.service.spark.tenant</name>
<value>1c5c9e97c8db488baeca8d667497aef7</value>
</property>
<property>
<name>fs.swift2d.service.spark.password</name>
<value>passw0rd</value>
</property>
<property>
<name>fs.swift2d.service.spark.username</name>
<value>d2a2adb8bd924c2da1545e2e9ee7c4fe</value>
</property>
<property>
<name>fs.swift2d.service.spark.auth.method</name>
<value>keystoneV3</value>
</property>
<property>
<name>fs.swift2d.service.spark.region</name>
<value>IBMCOS</value>
</property>
</configuration> |
1.将 your.keystone.server.com 替换为
Keystone 服务器的真实地址。
2.将所有身份验证凭证（tenant、username 和 password）替换为您的对象存储的有效凭证。请记住，对于
Keystone Version 3，使用 userID 和 tenantID 代替 username
和 tenant 值。
3.使用全球 Keystone 时，需要根据为 IBM COS
访问定义的 Keystone 地区来定义 region 属性。
在代码中以编程方式指定参数
如果喜欢在代码中指定参数，可以使用下面的代码示例，其中的 SERVICE_NAME 为 spark。
清单 4. 向代码添加配置参数
| hconf
= sc._jsc.hadoopConfiguration()
hconf.set("fs.swift2d.impl", "com.ibm.stocator
.fs.ObjectStoreFileSystem")
hconf.set("fs.swift2d.service.spark.auth.url",
"http://your.authentication.server.com/v2.0/tokens")
hconf.set("fs.swift2d.service.spark.public",
"true")
hconf.set("fs.swift2d.service.spark.tenant",
"service")
hconf.set("fs.swift2d.service.spark.username",
"swift")
hconf.set("fs.swift2d.service.spark.auth.method",
"keystone")
hconf.set("fs.swift2d.service.spark.password",
"passw0rd")
hconf.set("fs.swift2d.service.spark.region",
"IBMCOS") |
表 1 对每个参数进行了说明。
表 1. 配置参数
启动启用了 Stocator 的 Spark
在利用 Stocator 从 Spark 访问 IBM COS 对象之前，需要静态地将 Spark
和 Stocator 重新编译到一起，或者动态地将 Stocator 的库传递给 Spark。
要利用源代码重新编译 Spark 来包含 Stocator 驱动程序，请参阅 github 上的 Stocator
存储库 中的说明。
要通过 Stocator 的独立 jar 库使用 Spark，而不重新编译它，可以使用 –jars
选项运行 Spark。在我们的环境中，我们使用了 1.0.8 版的 Stocator，所以独立 jar
库的名称为 stocator-1.0.8-SNAPSHOT-jar-with-dependencies.jar。在我们的示例环境中，传递给
Spark 的选项包括：–jars stocator-1.0.8-SNAPSHOT-jar-with-dependencies.jar。
| ofer@beginnings:~$
~/spark-2.0.1/bin/spark-shell \
--jars stocator-1.0.8-SNAPSHOT-jar-with-dependencies.jar
Welcome to
____ __
/ __/__ ___ _____/ /__
_\ \/ _ \/ _ `/ __/ '_/
/___/ .__/\_,_/_/ /_/\_\ version 2.0.1
/_/
Using Scala version 2.11.8 (OpenJDK 64-Bit Server
VM, Java 1.7.0_111)
Type in expressions to have them evaluated.
Type :help for more information.
scala> |
从 Spark 访问 IBM COS 对象
在 Spark 上启用 Stocator 后，就可以使用模式 swift2d://<container>.<service>/
从 Spark 访问 IBM COS 对象。swift2d 关键字告诉 Spark 使用哪个驱动程序来访问存储。它表明您正在使用
Stocator 访问一个对象存储。容器和服务将在下一节中更详细地介绍。
例如，以下 Python 代码从 IBM COS 读取一个名为 data.json 的 JSON 对象，并将它作为一个名为
data.parquet 的 Parquet 对象写回。
清单 6. 访问 IBM COS 对象
| df
= sqlContext.read.json("swift2d://vault.spark/data.json”)
df.write.parquet("swift2d://vault.spark/data.parquet”) |
测试 Spark 与 IBM COS 之间的连接
为了测试 Spark 与 IBM COS 之间的连接，我们使用了一段简单的 Python 脚本，该脚本将单一列表
的 6 个元素分布在 Spark 集群上，将数据写入 Parquet 对象中，最后读回该对象。Parquet
对象的名称被作为参数传入脚本中。
该数据显示了两次：第一次是在写入对象存储之前，与其模式一起显示；第二次是在从对象存储读回之后。
清单 7. 测试连接的 Python 脚本
| from
pyspark import SparkContext
from pyspark.sql import SQLContext
from pyspark.sql.types import *
import sys
sc = SparkContext()
sqlContext = SQLContext(sc)
if (len(sys.argv) != 2):
print "ERROR: This program takes object
name as input"
sys.exit(0)
objectName = sys.argv[1]
myList = [[1,'a'],[2,'b'],[3,'c'],[4,'d'],[5,'e'],[6,'f']]
parallelList = sc.parallelize(myList).collect()
schema = StructType([StructField('column1',
IntegerType(), False),
StructField('column2', StringType(), False)])
df = sqlContext.createDataFrame(parallelList,
schema)
df.printSchema()
df.show()
dfTarget = df.coalesce(1)
dfTarget.write.parquet("swift2d://vault.spark/"
+ objectName)
dfRead = sqlContext.read.parquet("swift2d://vault.spark/"
+ objectName)
dfRead.show()
print "Done!" |
要运行该脚本，请完成以下步骤：
1.将代码以文件 sniff.test.py 的形式保存在 清单
7 中。
2.创建一个名为 vault 的容器。
3.将服务（url 中的容器名称后显示的词语）设置为 core-site.xml
文件中定义的 SERVICE_NAME（请记住，我们的示例中使用了 spark）。
4.发出以下命令，其中的 testing.parquet 是要创建并读取的对象的名称：spark-submit
--jars stocator-1.0.8-SNAPSHOT-jar-with-dependencies.jar
sniff.test.py testing.parquet。
您会在 IBM COS 中看到一个 testing.parquet 对象，以及以下 Spark 输出：
清单 8. 确认已连接的 Spark 结果
| root
|-- column1: integer (nullable = false)
|-- column2: string (nullable = false)
+-------+-------+
|column1|column2|
+-------+-------+
| 1| a|
| 2| b|
| 3| c|
| 4| d|
| 5| e|
| 6| f|
+-------+-------+
+-------+-------+
|column1|column2|
+-------+-------+
| 1| a|
| 2| b|
| 3| c|
| 4| d|
| 5| e|
| 6| f|
+-------+-------+
Done! |
结束语
通过配置 Spark、Stocator 和 IBM Cloud Object Storage 来协同工作，可以使用对象存储语义更快地访问和分析存储的数据，而无需使用为处理文件系统设计的旧式存储连接器。
|
