Learning Spark 中文版 -- 第五章 -- 加载保存数据（2）

File Compression(文件压缩)

  当我们处理大数据时，我们经常发现需要压缩数据来节省存储空间和网络开销。对于大多数大数据格式，我们可以制定压缩编码器来压缩数据。我们之前已经见识到，Spark 原生的输入格式可以自动为我们处理压缩类型。当你读取压缩文件时，有些压缩编码器可以自动判断压缩类型。

  这些压缩选项仅适用于支持压缩的 Hadoop 格式，即写入文件系统的格式。数据库 Hadoop 格式通常不支持压缩，或者它们在数据库自身配置了对数据的压缩。

  选择一个输出压缩编码器可能会对以后的数据使用人员造成很大的影响。像 Spark 这种分布式系统，我们通常会从多个不同的机器中读取数据。为了满足分布式的需求，每个工作节点需要找到新数据记录的开始点。有些压缩格式让人无从下手，因为如果需要单个节点读取数据，就很容易导致性能瓶颈。能够很容易从多个机器中读取的数据叫做“可分割性（splitable）”。表 5-3 罗列了可用的压缩选择。

Spark 的 textFile() 方法可以处理压缩的输入，即使在输入被压缩以致其可以以分割的形式读取的情况下，它也会自动禁用可分割性。如果你发现你需要读取一个很大的单个压缩输入数据，最好跳过 Spark 的包装，使用 newAPIHadoopFile 或者 hadoopFIle 并且明确指定正确的压缩编码器。

  有些输入格式（如 SequenceFile)允许我们压缩键值对数据的值，这样可以在查找时提高效率。其他输入格式有自己对压缩控制的方式：例如，推特的 Elephant Bird 包中的许多格式使用 LZO 压缩数据。

Filesystems（文件系统）

  Spark 支持很多文件系统的读取写出，我们可以使用我们想用的各种文件格式。

Local/"Regular" FS(本地或常规文件系统)

  尽管 Spark 支持从本地文件系统加载文件，但需要文件在集群所有节点的路径都相同才能够使用。

  一些网络文件系统，如 NFS，AFS 和 MapR 的 NFS 层，作为常规文件系统向用户公开。如果你的数据已经在上述某个文件系统中，指定一个路径

Example 5-29. Loading a compressed text file from the local filesystem in Scala

val rdd = sc.textFile("file:///home/holden/happypandas.gz")

  如果文件还没有在集群的所有节点上，你可以不同通过 Spark 在驱动程序上本地加载，然后调用 parallelize 将内容分发给 worker。这种方式速度很慢，所以我们建议把文件放在 HDFS,NFS 或 S3 等共享文件系统中。

Amazon S3(亚马逊的 s3)

  Amazon S3 是一个日渐流行的存储大数据的选择。当您的计算节点位于 Amazon EC2 内时，S3 非常快速，但是如果你必须在公网上传输，那可能会有很糟糕的表现。

  要访问 Spark 中的 S3，您应该首先将 AWS_ACCESS_KEY_ID 和 AWS_SECRET_ACCESS_KEY 环境变量设置为您的 S3 凭证。你可以从 Amazon Web Service 控制台创建这些凭证。然后传递一个以 s3n://开头的路径给 Spark 的读取文件方法，格式为 s3n://bucket/pathwithin-bucket。和其他文件系统一样，Spark 支持 S3 的通配符路径，如

  如果你从 Amazon 获得 S3 访问权限错误，请确保为其指定了访问密钥的帐户同时具有“read”和“list”权限。Spark 需要能够列出存储 bucket 中的对象以识别您想要读取的对象。

HDFS

  HDFS 是非常流行的分布式文件系统，Spark 与其对接的效果也很好。HDFS 被设计为在商用硬件上工作的系统，并且在提供高吞吐量的情况下还能保证对故障处理的弹性。Spark 和 HDFS 可以布置在同一台机器上，并且 Spark 可以借此避免网络通信开销。

  对输入输出的数据设置 port/path 就可以在 Spark 上使用 HDFS 了，非常简单。

HDFS 的协议在 Hadoop 不同版本之间存在差异，所以如果你运行一个针对不同版本编译的 Spark 版本，会导致失败。默认情况下，Spark 是针对
Hadoop 1.0.4 构建的。如果从源代码构建，则可以设置 SPARK_HADOOP_VERSION =指定为环境变量，以针对不同的版本构建；或者你可以下载一个不同的 Spark 预编译版本。你可以在运行时决定 hadoop 的版本。

Structured Data with Spark SQL(Spark SQL 处理结构化数据)

  Spark SQL 是 Spark1.0 加入的组件，很快变成 Spark 处理结构化和半结构化数据最受欢迎的方式。对于结构化数据，我们意思是固定模式的数据，就是数据记录之间具有一致的字段集。Spark SQL 支持多个结构化数据源作为输入，并且因其能够理解他们的格式，所以可以高效地读取数据源必须的字段。第九章我们会详细介绍 Spark SQL，现在我们会简要介绍少量常用数据源的使用。

  在所有的例子中，我们让 Spark SQL 在数据源上运行一个 SQL 查询（选择一些字段或字段的函数），并且会返回给我们 Row 对象的 RDD，每条记录一个。在 Java 和 Scala 中，允许 Row 对象基于列号进行访问。每个 Row 象都有一个 get() 方法，用来返回我们可以转换的一般类型，并且有对于常见的基本类型特殊的 get() 方法（如， getFloat(), getInt(), getLong(), getString(), getShort(), and getBoolean())。在 Python 中我们直接使用 row[column_number] 和 row.column_name 访问元素。

Apache Hive（Apache 的 Hive）

  Hadoop 中常用的一个结构数据就是 Apache 的 Hive。Hive 可以以各种格式存储表格，在 HDFS 或这其他存储系统中，从纯文本到列向格式的格式都可以。Spark SQL 可以支持任何 Hive 支持的表格。

  要将 Spark SQL 连接现有的 Hive，你需要提供 Hive 的配置。你可以通过复制你的 hive-site.xml 到 Spark 的./conf/directory 中进行连接。一旦你这样做了，你就创建了个 HiveContext 对象，它是 Spark SQL 的入口点，你可以写 Hive 的查询语句(HQL)来查询数据表，会返回一个行 RDD。示例如下：

Example 5-30. Creating a HiveContext and selecting data in Python

from pyspark.sql import HiveContext

hiveCtx = HiveContext(sc)
rows = hiveCtx.sql("SELECT name, age FROM users")
firstRow = rows.first()
print firstRow.name

Example 5-31. Creating a HiveContext and selecting data in Scala

import org.apache.spark.sql.hive.HiveContext

val hiveCtx = new org.apache.spark.sql.hive.HiveContext(sc)
val rows = hiveCtx.sql("SELECT name, age FROM users")
val firstRow = rows.first()
println(firstRow.getString(0)) // Field 0 is the name

Example 5-32. Creating a HiveContext and selecting data in Java

import org.apache.spark.sql.hive.HiveContext;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SchemaRDD;

HiveContext hiveCtx = new HiveContext(sc);
SchemaRDD rows = hiveCtx.sql("SELECT name, age FROM users");
Row firstRow = rows.first();
System.out.println(firstRow.getString(0)); // Field 0 is the name

170 页“Apache Hive”会详细介绍 Hive 如何加载数据。

JSON

  如果你数据记录的 JSON 数据结构一致，Spark SQL 可以推断它们的结构并且加载数据为行，使你可以非常容易地获取需要的字段。为了加载 JSON 数据，首先在需要使用 Hive 的时候创建一个 HiveContext 对象。（在这种情况下，不需要安装 Hive，也就是说，您不需要 hivesite.xml 文件。）然后使用 HiveContext.jsonFIle 方法来获得整个文件的 Row 对象 RDD。除了使用整个 Row 对象之外，你也可以将这个 RDD 注册为一个表并从中搜索特定字段。举例来讲，假如我们有一个 JSON 文件格式如下，并且每行一条数据：

Example 5-33. Sample tweets in JSON
{"user": {"name": "Holden", "location": "San Francisco"}, "text": "Nice day out today"}
{"user": {"name": "Matei", "location": "Berkeley"}, "text": "Even nicer here :)"}

  我们可以加载这段数据并且查询 username 和 text 字段，示例如下：

Example 5-34. JSON loading with Spark SQL in Python

tweets = hiveCtx.jsonFile("tweets.json")
tweets.registerTempTable("tweets")
results = hiveCtx.sql("SELECT user.name, text FROM tweets")

Example 5-35. JSON loading with Spark SQL in Scala

val tweets = hiveCtx.jsonFile("tweets.json")
tweets.registerTempTable("tweets")
val results = hiveCtx.sql("SELECT user.name, text FROM tweets")

Example 5-36. JSON loading with Spark SQL in Java

SchemaRDD tweets = hiveCtx.jsonFile(jsonFile);
tweets.registerTempTable("tweets");
SchemaRDD results = hiveCtx.sql("SELECT user.name, text FROM tweets");

  我们会在 172 页的“JSON”中详细讨论如何使用 Spark SQL 加载 JSON 数据和访问其模式。另外，
Spark SQL 支持的数据不仅仅是加载数据，包括查询数据，以更复杂的方式与 RDD 结合,以及在其上运行自定义函数，我们将在第 9 章中介绍。

Databases (数据库)

  Spark 可以使用他们的 Hadoop 连接器或自定义的 Spark 连接器访问多种流行的数据库。 在本节中，我们将展示四个通用连接器。

Java Database Connectivity（Java 数据库连接）

  Spark 可以从任何支持 Java 数据库连接（JDBC）的关系型数据库加载数据，包括 MySQL, Postgres 和其他系统。为了访问这些数据，我们构建一个 org.apache.spark.rdd.JdbcRDD 并为其提供了我们的 SparkContext 和其他参数。Example5-37 展示了使用 JdbcRDD 连接 MySQL 数据库。

Example 5-37. JdbcRDD in Scala

def createConnection() = {
    Class.forName("com.mysql.jdbc.Driver").newInstance();
    DriverManager.getConnection("jdbc:mysql://localhost/test?user=holden");
}

def extractValues(r: ResultSet) = {
    (r.getInt(1), r.getString(2))
}

val data = new JdbcRDD(sc,
    createConnection, "SELECT * FROM panda WHERE ? <= id AND id <= ?",
    lowerBound = 1, upperBound = 3, numPartitions = 2, mapRow = extractValues)
println(data.collect().toList)

  JdbcRDD 接收的几个参数:

• 首先，我们提供一个函数来建立数据库连接。这使得每个节点在执行任何连接所需的配置后，都可以创建自己的连接来加载数据。
• 然后我们提供一个读取一定范围数据的查询，以及此查询参数的上限和下限。这些参数允许 Spark 在不同机器上查询不同范围的数据，所以我们不会因在单个节点上加载所有数据而产生性能瓶颈。（如果您不知道有多少记录，则可以先手动进行计数查询并使用其结果来确定 upperBound 和 lowerBound。）
• 最后一个参数是一个函数，用来将 java.sql.ResultSet 的每行输出转换为对操作数据有用的格式。在 Example5-37 中，我们会拿到（Int，String）键值对。如果省略此参数，Spark 会自动将每行转换为一个对象数组。

  如其他的数据源，当使用 JdbcRDD 时，确保你的数据库可以处理从 Spark 并行读取的负载。如果你想离线查询数据而不是实时的数据库，你可以使用数据库的导出功能导出文本文件。

Cassandra

  随着从 DataStax 引入开源 Spark Cassandra 连接器，Spark 对 Cassandra 的支持已经大大改善。由于连接器目前不是 Spark 的一部分，所以你需要在项目中引入额外的依赖。Cassandra 目前还不能使用 Spark SQL，但是可以返回 CassandraRow 对象，它和 Spark SQL 的 Row 对象具有一些相同的方法，Eample5-38 和 5-39 展示了用法。Spark Cassandra 连接器目前只能在 Java 和 Scala 中使用。

Example 5-38. sbt requirements for Cassandra connector
//sbt配置
"com.datastax.spark" %% "spark-cassandra-connector" % "1.0.0-rc5",
"com.datastax.spark" %% "spark-cassandra-connector-java" % "1.0.0-rc5"

Example 5-39. Maven requirements for Cassandra connector
//maven配置
<dependency> <!-- Cassandra -->
    <groupId>com.datastax.spark</groupId>
    <artifactId>spark-cassandra-connector</artifactId>
    <version>1.0.0-rc5</version>
</dependency>
<dependency> <!-- Cassandra -->
    <groupId>com.datastax.spark</groupId>
    <artifactId>spark-cassandra-connector-java</artifactId>
    <version>1.0.0-rc5</version>
</dependency>

  很像 Elasticsearch，Cassandra 连接器读取一个 job 的属性然后确定连接哪个集群。通过设置 spark.cassandra.connection.host 来连接 Cassandra 集群并且如果有用户名和密码的话可以通过 spark.cassandra.auth.username 和 spark.cassandra.auth.password 设置。假如你只有一个可供连接的 Cassandra 集群，我们可以在创建 SparkContext 的时候设置。示例如下:

Example 5-40. Setting the Cassandra property in Scala

val conf = new SparkConf(true)
        .set("spark.cassandra.connection.host", "hostname")

val sc = new SparkContext(conf)


Example 5-41. Setting the Cassandra property in Java

SparkConf conf = new SparkConf(true)
    .set("spark.cassandra.connection.host", cassandraHost);
JavaSparkContext sc = new JavaSparkContext(
    sparkMaster, "basicquerycassandra", conf);

  Datastax Cassandra 连接器使用 Scala 的隐式转换在 SparkContext 和 RDD 上提供额外的函数，来引入一些隐式转换并加载一些数据（Example5-42）。

Example 5-42. Loading the entire table as an RDD with key/value data in Scala

// Implicits that add functions to the SparkContext & RDDs.
//导入隐式转换使用额外函数
import com.datastax.spark.connector._

// Read entire table as an RDD. Assumes your table test was created as
// CREATE TABLE test.kv(key text PRIMARY KEY, value int);
//引入整个表作为RDD。假设你创建的测试表结构是（key text PRIMARY KEY, value int）
val data = sc.cassandraTable("test" , "kv")
// Print some basic stats on the value field.
//打印对值字段的基本统计
data.map(row => row.getInt("value")).stats()

  在 Java 中没有隐式转换，所以我们需要显式地将我们的 SparkContext 和 RDD 转换为这个功能。示例如下:

Example 5-43. Loading the entire table as an RDD with key/value data in Java

import com.datastax.spark.connector.CassandraRow;
import static com.datastax.spark.connector.CassandraJavaUtil.javaFunctions;

// Read entire table as an RDD. Assumes your table test was created as
// CREATE TABLE test.kv(key text PRIMARY KEY, value int);
JavaRDD<CassandraRow> data = javaFunctions(sc).cassandraTable("test" , "kv");

// Print some basic stats.
System.out.println(data.mapToDouble(new DoubleFunction<CassandraRow>() {
    public double call(CassandraRow row) { return row.getInt("value"); }
}).stats());

  除了加载整个表，我们还可以查询数据的子集。我们可以通过在 cassandraTable（） 调用中添加 where 子句来限制数据，例如，sc.cassandraTable(…).where("key=?", "panda")。

  Cassandra 连接器支持保存不同类型的 RDD。我们可以直接保存 CassandraRow 对象 RDD，这在表之间复制数据很有用。通过指定列映射，我们可以将非行形式的 RDD 保存为元组和列表。示例如下:

Example 5-44. Saving to Cassandra in Scala

val rdd = sc.parallelize(List(Seq("moremagic", 1)))
rdd.saveToCassandra("test" , "kv", SomeColumns("key", "value"))

  这一部分只是很简要地介绍了 Cassandra 的连接器。如果想更深入地了解，可以查看连接器的 GitHub。

HBase

  Spakr 可以在实现了 org.apache.hadoop.hbase.mapreduce.TableInputFormat 的类中通过 Hadoop 的输入格式访问 HBase。这种输入格式返回键值对，键是 org.apache.hadoop.hbase.io.ImmutableBytesWritable 类型，值是 org.apache.hadoop.hbase.client.Result 类型。如 API 文档中所述，Result 类包含根据列族获取值的各种方法。

  为了在 Spark 中使用 HBase，你可以对正确的输入格式调用 SparkContext.newAPIHadoopRDD。Scala 示例如下:

Example 5-45. Scala example of reading from HBase

import org.apache.hadoop.hbase.HBaseConfiguration
import org.apache.hadoop.hbase.client.Result
import org.apache.hadoop.hbase.io.ImmutableBytesWritable
import org.apache.hadoop.hbase.mapreduce.TableInputFormat

val conf = HBaseConfiguration.create()
conf.set(TableInputFormat.INPUT_TABLE, "tablename") // which table to scan

val rdd = sc.newAPIHadoopRDD(
conf, classOf[TableInputFormat], classOf[ImmutableBytesWritable], classOf[Result])

  为了优化 HBase 的读取，TableInputFormat 包含多个设置，例如将扫描限制为只有一组列，并限制扫描的时间范围。你可以在 TableInputFormat API 文档中找到这些选项，并在将它们传递给 Spark 之前,对 HBaseConfiguration 设置。

Elasticsearch

  Spark 可以使用 Elasticsearch-Hadoop 从 Elasticsearch 读取和写入数据。Elasticsearch 是基于 Lucene 的开源搜索系统。

  Elasticsearch 连接器和我们之前介绍的连接器有些不同，因为它不需要提供路径信息，而是取决于 SparkContext 设置的配置信息。Elasticsearch 输出格式连接器也不需要使用 Spark 的包装，我们使用 saveAsHadoopDataSet，这意味着我们需要手动设置更多参数。让我们看看在 Elasticsearch 中读取写入简单数据的例子。

最新的 Elasticsearch Spark 连接器用起来很简单，支持返回 Spark SQL 行。之所以还介绍了这个连接器，因为行转换还不支持 Elasticsearch 中所有的原生类型。

Example 5-46. Elasticsearch output in Scala

val jobConf = new JobConf(sc.hadoopConfiguration)
jobConf.set("mapred.output.format.class", "org.elasticsearch.hadoop.mr.EsOutputFormat")
jobConf.setOutputCommitter(classOf[FileOutputCommitter])
jobConf.set(ConfigurationOptions.ES_RESOURCE_WRITE, "twitter/tweets")
jobConf.set(ConfigurationOptions.ES_NODES, "localhost")
FileOutputFormat.setOutputPath(jobConf, new Path("-"))
output.saveAsHadoopDataset(jobConf)

Example 5-47. Elasticsearch input in Scala

def mapWritableToInput(in: MapWritable): Map[String, String] = {
    in.map{case (k, v) => (k.toString, v.toString)}.toMap
}
val jobConf = new JobConf(sc.hadoopConfiguration)
jobConf.set(ConfigurationOptions.ES_RESOURCE_READ, args(1))
jobConf.set(ConfigurationOptions.ES_NODES, args(2))
val currentTweets = sc.hadoopRDD(jobConf,
    classOf[EsInputFormat[Object, MapWritable]], classOf[Object],
    classOf[MapWritable])
// Extract only the map

// Convert the MapWritable[Text, Text] to Map[String, String]
val tweets = currentTweets.map{ case (key, value) => mapWritableToInput(value) }

  与其他连接器相比，Elasticsearch 连接器稍微有些费解，但它对于如何使用这种类型的连接器提供了有用的参考。

在写出方面，Elasticsearch 可以进行映射推断，但这可能偶尔会错误推断，所以如果要存储除字符串以外的数据类型，则最好明确设置映射。

Conclusion（总结）

  随着本章的结尾，你现在应该能够将数据存入 Spark 中，并以对你有用的格式存储计算结果。我们分析了可用于数据的多种不同格式，以及压缩选项及其对使用数据方式的影响。随后的章节将研究如何编写更高效，更强大的 Spark 程序，以便我们可以加载和保存大型数据集。