在 EC2 上启动和运行 Shark

Question

可以使用 Spark 附带的 Spark EC2 脚本在 EC2 上启动 Shark 。这些脚本让您可以启动、暂停和销毁自动配置了 HDFS、Spark、Apache Mesos 和 Shark 的集群。

启动集群

要在 EC2 上运行 Shark 集群，首先在 Amazon Web Services site 上注册一个 Amazon EC2 帐户。然后，将 Spark 下载到本地计算机：

$ wget http://spark-project.org/files/spark-0.7.2-prebuilt-hadoop1.tgz
$ tar xvfz spark-0.7.2-prebuilt-hadoop1.tgz

ec2 目录包含用于设置集群的脚本。Spark EC2 指南 中提供了详细说明。简而言之，您需要执行以下操作：

$ spark-0.7.2/ec2/spark-ec2 -k <keypair-name> -i <key-file> -s <num-slaves> launch <cluster-name>

你的 EC2 密钥对的名字在哪里 <keypair>（你在创建它时给它的），<key-file> 是你的密钥对的私钥文件，<num-slaves> 是要启动的从节​​点的数量（首先尝试 1），<cluster-name> 是要启动的从节​​点的名称给你的集群。这将使用包含 Spark 和 Shark 的预构建机器映像在 EC2 上创建一个集群。

登录到主人使用 spark-ec2 login：

$ ./spark-ec2 -k key -i key.pem login <cluster-name>

然后，通过进入 shark 目录启动 Shark：

$ shark-0.2/bin/shark-withinfo

withinfo 脚本将 INFO 级别的日志消息打印到控制台。如果您愿意，您也可以通过运行 ./bin/shark.

访问 S3 中的数据

您可以使用 Hive 的 CREATE EXTERNAL TABLE 命令访问 S3 中目录中的数据。首先，通过将以下属性添加到中来配置您的 S3 凭据 ~/ephemeral-hdfs/conf/core-site.xml：

<property>
    <name>fs.s3n.awsAccessKeyId</name>
    <value>ID</value>
</property>
<property>
    <name>fs.s3n.awsSecretAccessKey</name>
    <value>SECRET</value>
</property>

在此之后你还应该：

export AWS_SECRET_ACCESS_KEY=awsSecretAccessKey
export AWS_ACCESS_KEY_ID=awsAccessKeyId

然后按照 Hive S3 指南 bin/shark 中的描述创建一个 S3 支持的表：

shark> CREATE EXTERNAL TABLE table_name (col1 type1, col2 type2, ...) <storage info> LOCATION 's3n://bucket/directory/';

在 HDFS 中创建表

spark-ec2 自动设置两个 HDFS 文件系统：ephemeral-hdfs，它使用附加到您的 VM 的临时磁盘，当 VM 停止时这些磁盘会消失，并且 persistent-hdfs，它由 EBS 支持并在暂停和启动同一集群时持续存在。默认情况下，Shark 将其表存储在中 ephemeral-hdfs，这提供了大量空间并且非常适合临时表，但不适合长期存储。您可以更改 HADOOP_HOME 以 conf/shark-env.sh 更改此设置，或将数据显式上传到 S3 或 persistent-hdfs.

与 Hive 一样，Shark 将其表存储在其 /user/hive/warehouse 配置的 HDFS 实例中。您可以在那里创建一个表 CREATE TABLE 并将数据上传到其中 /user/hive/warehouse/<table_name>，或者加载到 HDFS 中的其他地方并使用 CREATE EXTERNAL TABLE。

示例：维基百科数据

为了方便试用 Shark，我们提供了由 Freebase 收集的维基百科的小型和大型转储。它们位于 S3 目录 spark-data/wikipedia-sample 40 MB 和 spark-data/wikipedia-2010-09-12 50 GB 中。两者都存储为制表符分隔的文件，其中包含维基百科中每篇文章的一条记录，有五个字段：文章 ID、标题、修改日期、XML 和纯文本。

让我们首先为较小的样本数据集创建一个外部表：

shark> CREATE EXTERNAL TABLE wiki_small (id BIGINT, title STRING, last_modified STRING, xml STRING, text STRING)
       ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t' LOCATION 's3n://spark-data/wikipedia-sample/';

现在我们可以查询如下：

shark> SELECT COUNT(1) FROM wiki_small WHERE TEXT LIKE '%Berkeley%';
shark> SELECT title FROM wiki_small WHERE TEXT LIKE '%Berkeley%';

CREATE TABLE AS SELECT 我们还可以通过在表属性中启用“shark.cache”的语句将表缓存在内存中：

shark> CREATE TABLE wiki_small_in_mem TBLPROPERTIES ("shark.cache" = "true") AS SELECT * FROM wiki_small;

然后查询缓存数据以加快访问速度：

shark> SELECT COUNT(1) FROM wiki_small_in_mem WHERE TEXT LIKE '%Berkeley%';

或者，我们可以只缓存表的一个子集，例如只缓存其中的两列（或我们希望的任何其他 SQL 表达式）：

shark> CREATE TABLE title_and_text_in_mem TBLPROPERTIES ("shark.cache" = "true") AS SELECT title, text FROM wiki_small;

最后，您可以使用以下命令对完整的 50 GB 数据集尝试相同的命令：

shark> CREATE EXTERNAL TABLE wiki_full (id BIGINT, title STRING, last_modified STRING, xml STRING, text STRING)
       ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t' LOCATION 's3n://spark-data/wikipedia-2010-09-12/';

要快速处理这个更大的数据集，您的集群中可能需要至少 15 个 m1.xlarge EC2 节点。（例如传递 -s 15 -t m1.xlarge 给 spark-ec2。）在我们的测试中，使用这些设置启动的 15 节点集群可以在大约 80 秒内从 S3 扫描数据集，并且可以轻松地将 text 和 title 列缓存在内存中以将查询速度加快到大约 2 秒。