回到首页

输入关键词，多个关键词用空格分隔。

← 云计算课程 · 系列目录

云计算 第 14 课 文件 vs 数据库

约 27 分钟
本页目录

从这一课起，我们要开始使用数据库了。通过数据库和文件的性能对比以及 SQL 与 NoSQL 的对比，学会选择最合适的技术。

学习目标

  1. 了解使用文件来存储信息的优势和劣势
  2. 增加使用 awk, grep 等命令修改文件的经验
  3. 了解使用数据库来存储信息的优势和劣势
  4. 了解 MySQL (SQL) 和 HBase (NoSQL) 的不同
  5. 学会如何把数据载入到数据库中(MySQL, HBase)
  6. 学会使用 JDBC 连接 MySQL
  7. 学会使用 Java API 来操作 HBase
  8. 了解 vertical scaling 的在持久云存储（磁盘, 固态硬盘）的性能

这次的作业主要用 Bash 和 Java(MySQL & HBase) 在 AWS 平台上完成。

背景知识

近年来『数据』越来越被重视，这之中很重要的一环就是——如何存储这些数据。这一课中我们会接触常见的存储数据的方式，并学会在实际场景中根据需要选择合适的技术。

我们先会介绍『文件』以及『关系型数据库』

通常来说，我们用文件来保存非结构化的数据，用数据库来保存结构化的数据，我们来看看下面这个例子

# 文件中的一行
Name: Carnegie, Course: Cloud Computing, Section: A, Year: 2015

# 数据库中的一行（有四列）
Name           Course         Section     Year
Carnegie    Cloud Computing      A        2015

在数据库中，数据以表的形式存储，访问不同的元素比较简单，但是在文件中，就需要做一定的解析工作。文件和数据库各有所长，重要的还是具体问题具体分析，不能一概而论。

除了文件和传统的关系型数据库，NoSQL 数据库现在也越来越流行了。因为大数据面临的挑战，NoSQL 数据库在扩展性上比传统方法更好，但是却不得牺牲一些一致性和结构性来换取性能和可拓展性。

这节课我们同样会尝试在 HBase 上做一些操作。完成之后，应该能够对这三种方式有更加清晰的理解，以及能够根据实际使用场景来选择对应的方法。

背景设定

我们的目标是打造一个关于音乐和电影的社交网络，作为一个菜鸟，我拿到的第一个任务是分析音乐数据。在把服务部署到云上之前，公司希望我能评估一下用文件和用关系型数据库的性能比较。提供的数据文件如下：

其中 million_songs_metadata.csv 包含所有歌曲的信息，million_songs_sales_data.csv 包含一段时间内每首歌的每日销量。具体的格式如下：

Schema for file million_songs_metadata.csv

Schema for file million_songs_sales_data.csv

最后注意要给所有用到的资源打上 Project: 3.1 的标签

文件操作

这一部分主要是使用 grep 和 awk 来进行一些简单的数据处理工作，关于这两个命令的使用，本来是打算专门写日志来说明的（然而一直没抽出时间），所以就尽量在这里介绍得清晰一点。

Grep 命令可以用来查找文件中出现的关键词或者某种固定的模式，如果我们要找到一个文件中包含 "The Beatles" 的记录，那么可以用以下命令：

grep -P 'The Beatles' million_songs_metadata.csv

具体的查找过程可以有不同的参数进行设置，比方说下面的语句就会忽略大小写进行匹配：

grep -i -P 'The Beatles' million_songs_metadata.csv

关于不同参数的意义，可以直接在命令行中输入 man grep 进行查看。

利用管道，我们可以统计具体的行数，比如：

grep -P 'The Beatles' million_songs_metadata.csv | wc -l

用 grep 得到的结果，只要某一行出现了要找的内容，就算找到，但是如果我们想在指定的列中寻找特定的字符，就可以使用 awk 命令了。比方说，我们只想找出 artist_name 那一列中出现 "The Beatles" 的记录，就可以用下面的命令：

awk ' BEGIN {FS = ","} ; {if ($7 ~ /The Beatles/) { print; }}' million_songs_metadata.csv

这里 $7 表示是第 7 列，而 FS = "," 表示分隔符是 ,

如果我们想要更复杂一点的逻辑，比如要找到 Michael Jacksn 80 年代的歌曲，就可以用这个命令：

awk ' BEGIN {FS = ","} ; {if (tolower($7) ~ /michael jackson/ && $11 >= 1980 && $11 < 1990) { print; }}' million_songs_metadata.csv

随着问题越来越复杂，可能很多时候都没有办法在一行内解决问题，不过在这一部分，我们还是尽量试试看用 grep 和 awk 解决问题。

做好准备之后可以开启一个 ami-ca685ba0 的 t1.micro 实例来完成下面的任务。

基本我们要做的就是把 runner.sh 补充完整，仔细读题，仔细读题，仔细读题（比如是否区别大小写）。

第五题可以写一个程序或者若干命令，以 million_songs_metadata.csv 和 million_songs_sales_data.csv 中相同 track_id 为标准，合并两个文件。生成一个 million_songs_metadata_and_sales.csv 数据集，其中第 1 列是 track_id，第 2 列是 sales_date，第 3 列是 sales_count，第 4 - 13 列是 million_songs_metadata.csv 的其他列。

完成问题之后，可以使用 ./runner.sh files 来检查输出结果

提示

  1. 搜索找到一个能够完成合并文件的 unix 命令
  2. 只能用命令行脚本完成
  3. 不要使用 Java 和 Python
  4. 没有特别声明，所有的匹配都是大小写敏感的
  5. 第六题中，一个歌手可能有多个 artist_names，但是只会有一个唯一的 artist_id，应该根据 artist_id 来找到最大的销量，并返回所有 artist_name
  6. 注意保存好 runner.sh

解题攻略

首先先创建一个 ami-ca685ba0 的 t1.micro 实例。就绪之后 ssh 过去：ssh -i demo.pem ubuntu@ec2-54-175-177-74.compute-1.amazonaws.com，即可见到这次作业的相关文件：

这部分我们只需要用 runner.sh，所以把它搞到本地 scp -i demo.pem ubuntu@dns.compute-1.amazonaws.com:~/Project3_1/runner.sh ./

打开 runner.sh 文件，可以看到需要回答的问题是：

  1. 在文件 million_songs_metadata.csv 中，有多少行包含 Aerosmith，大小写敏感
  2. 在文件 million_songs_metadata.csv 中，artist_name 包含 Bob Marley 的 track_id 有多少个，大小写敏感
  3. 在文件 million_songs_metadata.csv 中，第 7 列中包含 The Beatles 的有多少行，大小写敏感
  4. 写出与 SQL 命令 SELECT AVG(duration) FROM songs 等价的命令行命令
  5. 把两个 csv 文件合并为 million_songs_metadata_and_sales.csv，以相同 track_id 为标准
  6. 在文件 million_songs_metadata_and_sales.csv 中，找到销量最高的 artist，一个歌手可能有多个 artist_names，但是只会有一个唯一的 artist_id，应该根据 artist_id 来找到最大的销量，并返回所有 artist_name

写好之后传到服务器上：scp -i demo.pem ./runner.sh ubuntu@dns.compute-1.amazonaws.com:~/Project3_1/

测试的话用 ./runner.sh files，确定无误后使用 ./submitter -a dawang 来进行提交，代码运行完成后输入提交密码即可。

MySQL 操作

同样是使用 ami-ca685ba0 的 t1.micro 实例来完成这部分的内容

先通过视频来了解 MySQL 的基础知识

远程机器中已经安装配置好了 MySQL，使用下面的命令可以开启 MySQL 命令行客户端并且连接到数据库：

mysql -u root -pdb15319root song_db

上面的命令中，用户名是 root 密码是 db15319root，使用是数据名称是 song_db

数据库的相关知识可以参考这里，就不在日志中赘述。

我们需要根据前面给出 schemas 来创建对应的表，作业文件中提供了 ~/Project3_1/create_tables.sql 文件，可以从这里开始

远程主机中的 MySQL 版本是 5.5，注意查看对应的文档

创建好之后，可以使用下面的命令来查看表的 schema

DESCRIBE songs;
DESCRIBE sales;

然后和前面给出的表格进行比较，看看是否一致。

所以我要做的是找到合适的命令，把 million_songs_metadata.csv 和 million_songs_sales.csv 导入到 MySQL 中。可以在 MySQL 命令行工具中使用 SQL 命令导入，也可以用 mysqlimport 工具来导入，记下所使用的命令即可。

想要验证是否导入成功的话，可以列出前十条记录：

SELECT * FROM songs 
LIMIT 10;

SQL 的语法可以参考这里，下面选出一些简单的例子进行介绍。

比如说下面的命令就会从表中选出 artist_name 一列中包含 The Beatles 的表项：

SELECT * FROM songs 
WHERE artist_name 

LIKE '%The Beatles%';

这里的 % 表示任何字符出现任意次数，前面提到的寻找 Michael Jackson 的例子可以写成：

SELECT * FROM songs 
WHERE artist_name 
LIKE '%michael jackson%'AND year >= 1980 AND year < 1990;

如果需要计算平均时间，就不需要使用 awk 命令那么复杂，可以直接

SELECT AVG(duration) FROM songs;

带索引的数据库可以极大提高查询的性能，在 MySQL 中，所有的主键都会自动成为索引。

Aggregate Functions 允许你在多个记录中执行运算并返回一个单一值，比较常用的有 SUM, AVG, MAX, MIN 和 COUNT. Aggregate functions 通常和 MySQL 的 GROUP BY 关键字一起使用来为不同的 subgroup 执行运算并返回对应结果。GROUP BY 非常有用，下面是一个例子：

SELECT c1, c2, ... cn, aggregate_function(expression)
FROM table
WHERE where_conditions
GROUP BY c1, c2, ... cn;

例如，要统计最近十天总销量排名，可以用下面的 SQL 语句：

SELECT sales_date, SUM(sales_count) AS total_sales
FROM sales
GROUP BY sales_date
ORDER BY sales_date DESC
LIMIT 10;

MySQL 的 JOIN 关键字可以用来在两个或两个以上相关的表中进行查询。在 MySQL 中 JOIN, CROSS JOIN 和 INNER JOIN 是等价的，下面是一个例子：syntax:

#select statement
    SELECT c1,c2,....cn
    FROM join_table;
#join_table
    table1 [INNER|CROSS] JOIN table2 [join_condition]
#join_condition:
    ON conditional_expr
  | USING (column_list)

INNER JOIN 会构造指定的表的笛卡尔乘积，也就是第一个表中的每一行通过 join condition 和第二个表中的每一行组合。因为我们的 songs 表中的所有 track_ids 在 sales 表中都有对应的记录，所以这里只用 INNER JOIN 即可。

例如，下面的 SQL 语句会返回销量最高的 10 首歌的名字：

SELECT songs.title, SUM(sales_count) AS total_sale
FROM songs
INNER JOIN sales ON songs.track_id = sales.track_id
GROUP BY sales.track_id
ORDER BY total_sale
DESC LIMIT 10;

关于 OUTER JOIN

和 INNER JOIN 不同的是可能会出现列的值为空的情况，根据提供不匹配的数据的表所在的位置，分为 LEFT 和 RIGHT JOINS。在 LEFT JOIN 中，会返回左边表中不匹配的记录，反之亦然。没有匹配的话，会把对应的列设为 NULL。如果想要保留不匹配的数据，这种方法就很有用了。

JDBC 和 MySQL

Java Database Connectivity (JDBC) API 可以用来访问数据库，并且由于是一个跨平台的标准，在不同的平台上可以使用相同的代码。这一部分我们会使用 MySQL Connector/J。

第一步就是与数据建立连接，如下：

Class.forName("com.mysql.jdbc.Driver");
Connection conn = DriverManager.getConnection(URL, DB_USER, DB_PWD);

第一行载入并初始化 MySQL 的 JDBC 驱动，然后我们就可以建立与数据库的连接（参数比较简单这里略过）

为了执行 SQL 操作以及获取执行完毕的结果，我们需要创建 Statement(用来执行 SQL 命令的对象)，并且在执行完成后得到一个 ResultSet 对象，下面是一个例子：

Statement stmt = conn.createStatement();
ResultSet rs = stmt.executeQuery("select count(*) as cnt from songs;");

可以通过调用 rs.next() 来遍历结果：

while (rs.next()) {
    int rowCount = rs.getInt("cnt");
    System.out.println("Total number of lines in songs is " + rowCount);
}

执行完对应操作后，还需要关闭我们用到的 Statement 和 Connection，注意关闭 Statement 之后对应的 ResultSet 也会被关闭。

解题攻略

和之前一样，我们要做的就是完成 7-11 题，需要修改 MySQLTasks.java 文件。仔细读题，仔细读题，仔细读题。

第 7-9 题我们需要为 songs 表创建索引。那么应该选择哪一列作为索引呢？

记录下使用的命令已经对应更新 INDEX_NAME 变量，建立索引需要花一点时间，不过可以换取比较大的性能提升，建立完索引后，使用下面的命令重启 mysql：

sudo service mysql restart

在第 9 题中，我们会使用和第 7 题一样的指令，就可以看到建立索引之后的性能提升。

开始写代码之前，可以先运行一下样例，了解 java 如何和 MySQL 交互。

javac MySQLTasks.java
java MySQLTasks demo

会输出 songs 表中的行数（如果存在的话），做完之后可以用 ./runner.sh mysql 来检查。

Bonus

如果完成 MySQLTasks.java 中的 loadData 函数，有 5 分的加分，可以通过下面代码进行测试：

javac MySQLTasks.java
java MySQLTasks load_data

一些提示

  • SQL 的 LIKE 操作符默认是大小写不敏感的
  • 记得把所有的答案输出到同一行
  • 下一部分也可以用同一个远程机器，终止之前保存好所有的代码

文件更灵活，可以存放结构或非结构数据，并且容易实现和修改；数据库则稍微笨重一些。对于文件来说，安全只能通过文件权限来控制，但是数据库有更加完善的权限管理。对文件的访问没有办法并行，但是数据库访问则可以。其他的不同基本上可以认为数据库有一套完整的管理接口和语法，而文件的话都需要自己实现，下表是一个总结：

文件 vs 数据库

首先我们把对应的文件复制到本地：scp -i demo.pem ubuntu@dns.compute-1.amazonaws.com:~/Project3_1/MySQLTasks.java ./

然后我们可以用给出的 create_tables.sql 来新建数据表，先进入 MySQL 的命令行：mysql -u root -pdb15319root song_db

然后输入 source ./create_tables.sql 来执行新建表格的命令。接着可以用 DESCRIBE songs; 和 DESCRIBE sales; 来查看是否成功创建（注意一定要最后的分号），如图

songs 表 sales 表

然后需要找到合适的命令，把 million_songs_metadata.csv 和 million_songs_sales.csv 导入到 MySQL 中（这里推荐用 mysqlimport 来导入，另一个有点问题）。命令为：

# 需要先更改两个 scv 文件的名字，这样才能载入到对应的表中
cp million_songs_metadata.csv songs.csv
cp million_songs_sales_data.csv sales.csv
# 然后进行载入
mysqlimport -u root -pdb15319root --local --fields-terminated-by="," --lines-terminated-by="\n" song_db songs.csv
mysqlimport -u root -pdb15319root --local --fields-terminated-by="," --lines-terminated-by="\n" song_db sales.csv

然后我们用下面的命令来看看是否成功（如果不成功，就重新用前面的脚本生成一次对应的表）：

SELECT * FROM songs LIMIT 10;
SELECT * FROM sales LIMIT 10;

大概应该看到

然后就可以进入写 java 代码的阶段了，先大概看一下已有的代码，发现已经帮我们初始化过了，实际上只要在代码中填写对应的 SQL 语句即可。问题为（第 7-11 题）：

  1. (7)返回 duration 最长的歌的 trackid
  2. (8)选择一列作为索引，并建立索引
  3. (9)返回 duration 最长的歌的 trackid（和第 1 个题目相同，用来比较性能）
  4. (10)写一条与 grep -P 'The Beatles' million_songs_metadata.csv | wc -l 等价的 sql 语句，这里注意大小写的问题，提示：BINARY（感谢 @jiexing）
  5. (11)哪个 artist 的歌曲数目是第三多的，返回其名字，如果有多个，任意一个都可以

其实主要就是写出对应的 SQL 语句，执行起来都是一样的，前面也有给出例子。然后就可以上传回服务器：scp -i demo.pem ./MySQLTasks.java ubuntu@dns.compute-1.amazonaws.com:~/Project3_1/

测试的话用 ./runner.sh mysql，确定无误后使用 ./submitter -a dawang 来进行提交，代码运行完成后输入提交密码即可。

Vertical Scaling 存储

那么问题来了，我们的数据到底保存在哪里呢？当然是物理世界的硬盘上，但是我们之前好像都没有考虑到这个事情，事实上，不同的硬盘对性能也有极大的影响。

接下来的部分我们会了解一些 Linux 下的磁盘操作命令并且利用 AWS 提供的存储设备来进行 vertical scaling。并且用常见的 benchmarking 工具来进行测试，通过整个过程，应该就能了解为什么实际存储数据的设备也对性能有极大的影响。

因为大多数命令都需要 root 权限，所以开始之前 sudo su 一下是比较方便的选择。

这个视频介绍如何在 EC2 实例中使用 EBS

  • 一般来说在创建 EC2 实例的时候会自动创建一个 EBS 并挂载到 EC2 实例上
  • 先进入 EBS Volume 页面，Create Volume -> 选择不同的大小 -> 选择 Availablility Zone(要和 EC2 在同一个区域) -> Create
  • 点击 Action -> Attach Volumn -> 选择已有的实例 -> 填写挂载点 /dev/sdf
  • ssh 到机器上，输入命令 sudo parted -l 可以发现并没有成功挂载
  • 我们在磁盘上新建一个文件系统：sudo mkfs.ext4 /dev/xvdf
  • 再次运行 sudo parted -l，发现一切正常
  • 然后创建文件夹用来挂载 sudo mkdir /mnt/ebs1
  • 接着进行挂载 sudo mount /dev/xvdf /mnt/ebs1
  • 就可以访问对应文件夹了 cd /mnt/ebs1/
  • 最后可以用 df -h 来进行查看

GNU parted 是用来创建、销毁、改变大小、检查状态、复制分区的命令，可以操作分区表（取代原来的 fdisk），并支持如 GUID Partition Table(GPT) 等的新特性。想要了解更多可以参考这里

parted -l
/dev/xvda1 – this is the OS partition
/dev/xvdb – this is the first Ephemeral (instance store) drive
/dev/xvdc – this is the second Ephemeral (instance store) drive

创建并格式化一个分区

umount /dev/xvdX #where “X” - is a,b,c..etc (You should use your device’s name)
parted /dev/xvdX mklabel gpt
parted /dev/xvdX mkpart db ext4 0% 10G
mkfs.ext4 /dev/xvdX1

对于比较小的 volume，可以直接整个格式化，不用分区

mkfs.ext4 /dev/xvdX

创建挂载点并挂载

mkdir /storage/mountpoint
mount /dev/yourdevice /storage/mountpoint

到底用不用挂载点可以自己决定，不过一般来说 Linux 会挂载到 /mnt（EC2 也是这么做的）

mount

不带任何参数的话会显示所有的挂载点，可以用来判断是否挂载成功。

解题攻略

简单来说就是比较传统硬盘和固态硬盘的性能差别，测试的场景如下：

Sysbench 是一个包含多个测试的评测。这里我们使用的评测程序和 sysbench 唯一不同的是可以选择 [SSD|Magnetic]。

根据下面的指示完成不同配置的测试，记录下不同的 RPS，把数字填写到对应的位置即可。

提示：使用比较慢的硬件时，准备 10GB 的数据可能要花费很长时间，用最好的的机器（比如 large）来准备数据

准备测试数据

步骤如下

  1. 用 ami-ca685ba0 启动一个 t1.micro 或 m3.large 的实例
  2. 创建一个 20GB 的 EBS volume (磁盘或固态硬盘) 。确保和 EC2 实例在同一个区域
  3. 把 EBS volume 挂载到 EC2 实例上
  4. SSH 到 EC2 实例，格式化并挂载 EBS volume
  5. 进入挂载文件夹
  6. 用下面的命令生成测试数据 sudo /home/ubuntu/Project3_1/sysbench --test=fileio --file-total-size=10G prepare

上面的命令会在 EBS volume 上生成 10GB 的测试数据，在接下来的步骤中都可以重复使用

试验 1 (上表中 Scenarios 1 & 2 )

执行以下步骤：

  1. 启动一个 ami-ca685ba0 的 t1.micro 实例
  2. 挂载上 EBS volume
  3. 确保挂载成功
  4. 执行下面的代码 3 次（中间不要间隔太长时间）sudo /home/ubuntu/Project3_1/sysbench --test=fileio --file-total-size=10G --file-test-mode=rndrw --max-time=300 --max-requests=0 run
  5. 把结果写到 runner.sh 中
  6. 使用另外的磁盘类型进行测试

暂时不要删除 EBS volume，之后还有用

试验 2 (上表中 Scenarios 3 & 4 )

  1. 启动一个 ami-ca685ba0 的 m3.large 实例
  2. 挂载上 EBS volume
  3. 确保挂载成功
  4. 执行下面的代码 3 次（中间不要间隔太长时间）sudo /home/ubuntu/Project3_1/sysbench --test=fileio --file-total-size=10G --file-test-mode=rndrw --max-time=300 --max-requests=0 run
  5. 把结果写到 runner.sh 中
  6. 使用另外的磁盘类型进行测试

测试的话用 ./runner.sh scaling，确定无误后使用 ./submitter -a dawang 来进行提交，代码运行完成后输入提交密码即可。

HBase 操作

Apache HBase 是一个开源版本的 Google BigTable 分布式存储系统，其特点是分布式，可拓展，高性能，为大数据而生，在 Hadoop Distributed File System (HDFS) 上工作。HBase 在不同的服务器上把文件保存为重复的块，HDFS 保证其扩展性和可靠性。

在 HBase 中，输入按照行列排列，如下图所示：

HBase table 的典型架构

HBase 中的每一行都有对应的 row key，类似于主键，必须是唯一的。HBase 会自动根据 row key 来排列数据，默认按照字节顺序排序。

如上图所示，每一行包括：rowkey, column_family, column 和 timestamp，所以整个的映射变成 (rowkey, column family, column, timestamp) -> value。Rowkey 和 value 都是简单的字节，所以只要能序列化成字节的都可以保存在 cell 中。这些 cell 会按照字典序排列，这是一个非常重要的特性，使得 HBase 支持快速搜索。

HBase 中的每一列都有列名，还可以进一步组织成 column family。所有的 column family 成员拥有共同的前缀，如上图所示，列 Metadata:Type 和列 Metadata:Language 都是 Metadata column family 的成员，而列 Content:Data 则属于 Content family。默认来说用冒号来分隔 column family 的前缀，这个前缀必须由能够打印的字符组成，后面的部分可以是任何字节。

HBase 操作

HBase 有四个主要的操作：Get, Put, Scan, 和 Delete.

  • Get 操作会返回指定行的所有 cell
  • Put 操作可以添加新的记录或者更新已有记录
  • Scan 操作会根据条件遍历多行记录
  • Delete 操作会移除一条记录

Get 和 Scan 操作的返回都是排好序的，依据为 rowkey, column family, family 成员，和时间戳（也就是最新的值会在最前面）。默认来说，Get, Scan 和 Delete 操作都是在数据最新的版本上的（也可以指定其他版本的数据）。Delete 操作一般来说会删除整行，但是也可以删除指定的 cell。

HBase 架构

HBase 是以 HBase 节点集群来进行组织的，节点有两种类型：master 和 slave（也叫 RegionServers）

HBase 集群架构

HBase 会动态分配数据表，这样支持大量的并行访问。一个 HBase 表在太大时会被分成多个 Region，一个 HBase Region 是一个 HBase 表的子集，但是 rowkey 的范围是连续的。每个 RegionServer 可以保存多个 Regions，但是一个 Region 只会在一个 RegionServer 上。

虽然一个 Region 只会在一个 RegionServer 上，但是这不意味着该 Region 部分的数据只能存在于一个 RegionServer 上。事实上，因为 HDFS 的复制机制，每个 Region 都会在其他 RegionServer 上有几份一模一样的拷贝。想要了解更多？查看 HBase Reference Guide 以及 HBase 博客.

HBase 使用 Apache ZooKeeper 来协调控制整个 HBase 集群。Apache ZooKeeper 需要做的事情有：选择 master 节点，寻找 -ROOT- catalog table 以及节点注册（当新的 RegionServer 加入的时候）。由 ZooKeeper 选择出来的 master 节点会处理诸如 region 分配，失败处理，负载均衡等任务。

HBase 使用 HDFS 作为存储，但是同样支持其他文件系统（本地文件系统，甚至 Amazon S3）。

这个 HBase Demo 视频会介绍 HBase 的基本使用，虽然视频中的 EMR 版本较旧，但是对我们这次的任务没有什么影响。

  • Costs = Instance + EMR costs
  • 进入 EMR 页面 -> 创建集群 -> 输入名字 -> 开启关闭保护
  • 选择 S3 的 log bucket -> 打上标签 -> 去掉 pig 和 hive，改为 HBase
  • 选择 spot -> 指定 keypair -> 需要等待一段时间开启
  • ssh 到 master public dns，注意这里用户名是 hadoop 而不是 ubuntu
  • hbase shell -> > help 查看帮助 -> > status 查看状态
  • create 'users', 'info' 创建表格
  • describe 'users' 可以查看表格内容
  • put 'user', 'johndeo', 'info', 'regularUser' 插入一条记录
  • get 'users', 'johndoe' 获取一条记录
  • scan 'users' 遍历某个表并输出
  • count 'users' 统计表的行数

注意：EMR 很贵，最好使用 spot instance

使用 EMR 创建 HBase 集群

我们将使用 EMR 创建 HBase 集群。HBase 使用 Hadoop Distributed File System (HDFS) 来存储数据。默认来说 AWS 会直接用 EC2 内置的存储给 HDFS 使用，下面是具体的使用步骤：

  1. 启动 EMR 集群：1 master & 1 core
    • 在创建页面中选择 "Go to advanced options"
    • 确保所有的实例都是 m1.large
    • 确保 EMR 集群和存放 runner.sh 的实例在同一个区域
    • 选择 AMI version 3.11.0 (hadoop version 2).
    • 移除所有的已有服务(Pig & Hive)并选择安装 HBase version 0.94.
    • 指定 key-pair 以便 SSH 到 master 实例，ssh 的时候注意用户名是 hadoop
    • 不要忘记设置标签
    • 开启 "termination protection" 和 "keep-alive"
  2. master 和 core 节点的安全组都允许所有流量，使用 Master public DNS 来进行连接
  3. ssh 到 master 节点之后，运行 hadoop dfsadmin -report 检查 HDFS 的状态

载入数据到 HBase

HBase 支持多种数据导入方法，这里我们介绍 Bulk Load 方法。

最直接的载入办法可以是在 MapReduce job 中使用 TableOutputFormat 类，也可以使用client APIs，但是这可能不是最有效率，因为 API 不支持 bulk loading.

Bulk Importing 会越过 HBase API 直接写入到数据文件中(HFiles)。使用 bulk load 可以减少 CPU 和网络带宽的占用。ImportTsv 就可以完成这个任务，虽然原本是为 TSV (Tab Separated Value) 格式设计的，但是通过设置参数，同样支持 CSV 文件，步骤如下：

  1. 把 TSV/CSV 格式的数据集上传到 HDFS (Hadoop Distributed File System)
    • File System (FS) shell 支持基本的文件操作比如 Local FS, HFTP FS, S3 FS 等等，可以通过 hadoop fs <args> 来调用
    • 从 S3 bucket 获取 million_songs_metadata.csv 文件
      • mkdir P3_1
      • cd P3_1
      • wget https://s3.amazonaws.com/15319-p31/million_songs_metadata.csv
    • 把下载下来的文件保存到 HDFS 中以便导入，具体命令需要自己寻找
    • 可以用 hadoop fs -ls /path/containing/your/uploaded/file 来检测是否上传成功
  2. 打开 HBase shell (HBase shell)并新建一个名为 songdata 的表(使用 create 命令，后面跟 column family name 的名字)。建立成功之后使用 exit 命令退出
  3. 为 HBase 表准备好 HFiles。使用 ImportTsv 命令把文件 million_songs_metadata.csv 中的数据传到 HDFS 中，名为 importtsv.bulk.outputHbase。这些 StoreFiles 之后会被载入到 HBase 中。注意这里我们使用 track_id 作为 row key，其他的列会成为 column family name (这里使用 ‘data’)。要了解 ImportTsv 的更多信息，请参考 official reference.
  4. 正常启动的话，我们可以看到 MapReduce 工作的进程
  5. 检查 Map 步骤的输出来验证结果。通常来说应该会与数据集中的数据数量相等。注意，对应的输出文件应该是不存在的（不然会导致任务失败）
  6. 前面所做的所有工作都只是为了把数据保存到 HBase 中，但是此时 HBase 的表仍旧是空的（还没有添加对应的记录）
  7. 需要使用 CompleteBulkLoad 工具来完成数据上传，参考官方文档来使用
  8. 现在可以验证数据是否成功上传，打开 HBase shell 然后用以下命令来查看 scan 'songdata'
  9. 用 Ctrl-C 结束输出

当然，除了这个方法，也可以在 MapReduce job 中使用 TableOutputFormat 或者其他 HBase client API。

HBase 查询

与 MySQL 类似，HBase 提供了查询的工具。在 HBase 中，数据存在 column 中，多个 column 组成 column family。我们可以用下面的指令来进行查询：

scan ‘table_name’, {COLUMNS => [‘column1’, ‘column2’, …], FILTER => “(FILTER1) … (FILTER2)”}

我们来做一个和之前类似的查询，找到所有 artist_name 以 “The Beatles” 开头的记录（是一个前缀匹配，不是子串匹配），查询如下：

scan 'songdata', {COLUMNS => 'data:artist_name', FILTER => "SingleColumnValueFilter('data', 'artist_name', = , 'regexstring:^The Beatles.*')"}

这里的列名的格式是 (column family name):(column qualifier name)。并且返回的数据中只包含了 artist_name 的数据，如果我们想多看一些数据，在 COLUMNS 部分多加一些内容，如：

scan 'songdata', {COLUMNS => ['data:artist_name', 'data:title'], FILTER => "SingleColumnValueFilter('data', 'artist_name', = , 'regexstring:^The Beatles.*')"}

同样，我们也可以添加更多的 FILTER，用逻辑运算符 AND, OR, WHILE 等来进行组合。比如说，如果我们想在原来条件的基础上增加另一个条件：其 title 以 W 或者以 W 之后的字母开头，那么命令就可以这么写：

scan 'songdata', {COLUMNS => ['data:artist_name', 'data:title'], FILTER => "SingleColumnValueFilter('data', 'artist_name', = , 'regexstring:^The Beatles.*') AND SingleColumnValueFilter('data', 'title', >= , 'binaryprefix:W')"}

另外提一点，在 FILTER 中使用了某一列，就需要在 COLUMNS 列表中也加入对应的列名，不然就会被忽略的，更多信息可以参阅这篇日志.

HBase Java API

HBase 也有其 Java API，可以用来创建、查看、修改和删除表，同样也可以插入和查询。

建立连接

首先我们需要建立连接，例如：

Configuration conf = HBaseConfiguration.create();
conf.set("hbase.zookeeper.quorum", zookeeperAddress);
conf.set("hbase.zookeeper.property.clientport", "2181");
HConnection conn = HConnectionManager.createConnection(conf);
HTableInterface table = conn.getTable(tableName);

前三行配置地址和端口，这里需要填写 master node 的 IP 地址。然后就可以创建 HConnection 并得到一个 HTableInterface 对象（用来处理特定 HBase 表）。

另一个创建 HBase table handler 的方法是

HTable table = new HTable(conf, tableName);

不过在新版本中已经被弃用了（所以直接不写出来不就好了嘛）

最常见的操作是 Get 和 Scan，get 用来获取某一行，scan 用来对多行操作，一般来说 scan 比 get 慢。不过我们这里会使用 scan。

下面是一个简单的例子，我们打印出所有 artist_name 以 "The Beatles" 开头的记录。更多详细的使用方法请参考 HBase Java API 文档.

// Create a new Scan object. By calling the default constructor, the entire table will be scanned.
Scan scan = new Scan();

// Binary representation of the column family name
byte[] bColFamily = Bytes.toBytes("data")

// Binary representation of the column name.
byte[] bCol = Bytes.toBytes("artist_name");

// This is used for regular expression matching. You should use different comparators based on specific requirements.
RegexStringComparator comp = new RegexStringComparator("^The Beatles.*");

// This defines the filtering rules of our Scan object.
Filter filter = new SingleColumnValueFilter(bColFamily, bCol, CompareFilter.CompareOp.EQUAL, comp);

// Associate the filtering rules to our Scan object.
scan.setFilter(filter);

// Use this if your query will return multiple rows.
scan.setBatch(10);

// Get the scan result.
ResultScanner rs = songsTable.getScanner(scan);

// Each call of rs.next() will return one row.
for (Result r = rs.next(); r != null; r = rs.next()) {
    // r represents one row in the table. r.getValue returns the specific cell (determined by column family
    // and column name.
    System.out.println(Bytes.toString(r.getValue(bColFamily, bCol)));
}

// Cleanup
rs.close();

看看 HBase tutorial on Client Request Filters 对完成这部分的任务也很有帮助。

解题攻略

这部分的任务就是完成 runner.sh 中的 17-21 题，需要改动的文件是 HBaseTasks.java。可以用下面的代码来运行 demo

javac HBaseTasks.java
java HBaseTasks demo

会打印出所有 artist_name 以 "The Beatles" 开头的记录（大小写敏感）

先把需要的文件 scp 到本地 scp -i demo.pem ubuntu@ec2-54-209-165-121.compute-1.amazonaws.com:~/Project3_1/HBaseTasks.java ./

问题列表(17 题开始)：

  1. (17)找到以 "Total" 开头 "Water 结尾的歌名
  2. (18)找到 "Kanye West" 的歌曲的歌名，名称以 "Apologies" 或 "Confessions" 开头，大小写敏感
  3. (19)找到歌手名以 "Bob Marley" 为前缀的一首歌的歌名，长度大于 400，年份是 2000 年之后（包括 2000 年）
  4. (20)找到歌手名包含 "Consequence" 的一首歌的歌名，歌名包含 "Family" 并且 artist_hotttnesss 要大于 1
  5. (21)找到歌手名以 "Gwen Guthrie" 为前缀的一首歌的歌名，歌名包含 "Love" 但不包含 "Bitter" 或者 "Never"，年份为 1990

然后按照前面的指引开一个 EMR，注意一定要开启 SSH，不然开了等于白开，开启之后连接上去 ssh -i demo.pem hadoop@ec2-52-90-21-43.compute-1.amazonaws.com

然后用 hadoop dfsadmin -report 检查状态，不过说已经弃用这种命令写法了，如下：

然后我们创建一个文件夹并下载对应的 csv 文件：

mkdir P3_1
cd P3_1
wget https://s3.amazonaws.com/15319-p31/million_songs_metadata.csv

然后创建对应的 HDFS 目录，再把 csv 文件移过去：

hadoop fs -mkdir /dawang
hadoop fs -mkdir /dawang/csv
hadoop fs -put ./million_songs_metadata.csv /dawang/csv/
# 查看
hadoop fs -ls /dawang/csv/

hadoop fs -ls 结果

然后进入 HBase Shell 操作 hbase shell

hbase(main):001:0> create 'songdata','data'
hbase(main):002:0> list
hbase(main):003:0> describe 'songdata'
hbase(main):004:0> exit

结果如下所示：

然后就需要具体的导入了，命令如下：

hbase org.apache.hadoop.hbase.mapreduce.ImportTsv -Dimporttsv.separator="," -Dimporttsv.bulk.output=/hfile_p31 -Dimporttsv.columns=HBASE_ROW_KEY,data:title,data:song_id,data:release,data:artist_id,data:artist_mbid,data:artist_name,data:duration,data:artist_familiarity,data:artist_hotttnesss,data:year songdata /dawang/csv/million_songs_metadata.csv

hbase org.apache.hadoop.hbase.mapreduce.LoadIncrementalHFiles /hfile_p31 songdata

完成之后测试一下 hbase shell：

hbase(main):001:0> scan 'songdata'

结果输出

有很多需要注意的地方，尤其是比较字符串的时候，有些坑是一定要踩的（爆炸感谢 @jiexing）。

测试的话用 ./runner.sh hbase，确定无误后使用 ./submitter -a dawang 来进行提交，代码运行完成后输入提交密码即可。

一些需要注意的地方：

  1. Java 代码中需要填写 HBase 的 master node 的 dns
  2. 每题的答案在一行里输出
  3. 设置正确的日志级别来防止不必要的输出

SCAN 操作是 O(N) 的，GET 操作是 O(logN)，比较好的方式是，通过精心设计的数据库，用两次 GET 操作拿到起始和结束的 rowkey，这样就有极大的效率提高，详情参考这里

参考资料

  1. Linux join命令

相关文章