当用户使用 MapReduce作业从表中读取数据时,实质上是使用 TableInputFormat取得数据。它符合 MapReduce框架,并重载了公有方法 getSplits()和 createRecordreader()。
在一个作业被执行前,框架调用 netSplit(来决定如何划分块(chunk),从而由块数目决定映射任务的数目。
对于 HBase来说, TableInputFormat基于用户提供的Scan实例取得所需要的表信息,并且按 region来划分边界。由于它不能直接地预测到可选过滤器的执行效果,所以它简单地使用起止键来确定 region。拆分块的数目与起止键之间的 region数目相等。如果用户没有设置这两个键,则所有 region都被包含在其中。
作业启动时,框架会按拆分的数目调用 createRecordReader(),并返回与当前块对应的TableRecordReader实例。换句话说,每个 TableRecordReader实例处理一个对应的regoin,读取并遍历一个 region的所有行每一个拆分也包含对应 region所在服务器的信息。正是这个信息能使 HBase上的MapReduce作业本地化执行:框架会比较 region对应的服务器名,如果有任务 tracker在对应服务器上运行,它会挑选这个服务器来执行作业。因为 region服务器与 Data Node运行在一个节点,所以扫描会从本地磁盘读取文件。
当在 HBase上运行 MapReduce时,推荐用户关闭预测执行模式。这样会增、加region和服务器的负载,同时与本地化运行相违背:预测作业在一个不同的机器上运行,因此没有本地 region服务器,这会增加网络带宽开销。
在 HBase之上的 MapReduce
1、准备
当运行 MapReduce业所需库中的类不是绑定在 Hadoop或 MapReduce框架中时,用户就必须确保这些库在作业执行之前已经可用。用户一般有两个选择:在所有的任务节点上准备静态的库,或直接提供作业所需的所有库。
1.1、静态配置
对于经常使用的库来说,最好将这些JAR文件安装在 MapReduce任务 tracker的本地,可以通过以下步骤来完成
1.1.1、把JAR文件复制到所有节点的常用路径中
1.1.2、将这些JAR文件的完整路径写入 hadoop-enνsh配置文件中,按照如下方式编辑 HADOOP_CLASSPATH变量:
1.1.3、重启所有的任务 tracker使变更生效。
显然这个技术是纯静态的,并且每次变更(即增加新库)都需要重启任务 tracker。如果需要添加 HBase的支持,至少需要增加 HBase和 ZooKeeper的JAR文件。按照如下方式编辑 hadoop-env.sh:
这里的前提是假设用户已经定义了两个$XYZ_HOME环境变量,并在环境变量中指出软件的安装路径。
1.2、动态配置
在一些情况下,用户希望每个他们运行的作业可以使用不同的库,或者在作业运行时与MapReduce集成能够更新库,这种情况下动态配置就非常有用了。
针对这种情况, Hadoop有一个特殊的功能:它可以读取操作目录中/lib目录下包含的所有库的JAR文件。用户可以使用此功能生成所谓的“胖”JAR文件,因为它们传输的不仅仅是实际的作业代码,还有所有需要的库。这意味着作业JAR文件更大,但从另一方面来说,其中已经自已包含了处理作业的完整的代码。
另一种动态提供必备库的方式是 Hadoop MapReduce框架的 libjars功能。当用户使用GenericoptionsParser创建一个 MapReduce作业时,可以得到libjar参数提供的支持。
HADOOP_CLASSPATH需要使用命令行。 Driver类在启动时需要访问 HBase和 Zookeeper类,需要按照如下方式添加libjars参数:
最终, HBase辅助类 TableMapReduceUti1提供了可以帮助用户在作业中通过代码动态添加依赖的JAR和配置文件的方法:
static void addDependencyJars(Job job) throws IOException;
static void addDependencyJars(Configuration conf, Class.. classes)throws IOException
用户过去都使用后一种功能来添加所需的 HBase、 ZooKeeper和作业类:
addDependencyJars (job.getConfiguration(),org.apache.zookeeper.ZooKeeper.class,
job.getMapoutputKeyClass(),
job.getMapoutput ValueClass(),
job.getInput Formatclass(),
job.getoutputKeyclass(),
job.getoutputValueClass(),
job.getoutputFormatclass(),
job.getPartitionerclass(),
job.getcombinerClass())
这里首先调用 addDependencyJars()方法添加作业和必要的类,包括输入输出格式和键值的类型等。第二个调用添加了Google Guava JAR,这个JAR需求程度在其他库之上。需要注意的是,这个方法并不需要用户指定实际的JAR文件,它直接使用Java的类加载器进行加载但有可能会找到相同的JAR文件,不过在这里这是无关紧要的,重要的是你可以在Java CLASSPATH中访问该类;否则,这些调用最终会失败,并返回一个 ClassNotFoundException错误,类似于之前你看到过的信息。用户在一个没有准备好的 Hadoop起动中,至少需要填加 HADOOP_CLASSPATH到命令行中。
你将选择哪种方法胖JAR的优势在于,其包含 HBase启动过程所有作业所需的库,而另一种方式则至少需要准备 classpath。
2、数据流向
随后,我们将使用 MapReduce作业作为过程的一部分,该作业可以使用 HBase进行读取或写入。第一个例子是使用 HBase作为数据流向,这个例子是在 TableoutputFormat:类的帮助下实现的。
MapReduce作业从一个文件中读取数据井写入 HBase表
- 为后续的使用定义一个作业名。
- 定义mapr类,继承自 Hadoop已有的类。
- map()函数将 Input Fort提供的键值对转化为了 OutputFormat需要的类型。
- 行键是经过MD5散列之后随机生成的键值。
- 存储原始数据到给定的表中的一列
- 使用 Apache Commons CLI类解析命令行参数。这些已经是HBae的一部分,因此用户可以很方便地处理作业的参数
- 传递命令行参数到解析器中,并优先解析“-Dxyz”属性。
- 使用特定的类定义作业。
- 这是一个只包含map阶段的作业,框架会直接跳过 reduce阶段。
用户在代码的main()函数中进行了设置,在运行 MapReduce作业之前,首先解析命令行以获取目标表名、目标列和输入文件的名称。虽然这里可以硬编码,但是最好还是写一段支持可配置的代码。
下一步是构建作业实例,通过命令行和已确定的参数(如类名)指定变量的细节。其中的一个细节是指定 mapper类,将其设置为ImportMapper。这个类与main类在同一个源代码文件中,并定义了作业在map阶段要完成的逻辑。
Main()函数的代码指定了输出格式的类,即前面描述过的 TableoutputFormat类。这个类由HBae提供,并能够方便地向一张表中写入数据,键的类型被该类隐式地设置为 ImmutableBytesWritable类,值被隐式地设置为 Writable的类。
用户在执行作业前,最好先创建一个目标表,例如,使用 HBase shel来创建目标表:
一旦准备好了表,用户就可以运行该作业了:
第一个命令是hadoop dfs-put,该命令将样本数据集合存储到HDFS的用户目录中。第二个命令则可以运行这个作业,作业的完成需要一段时间。使用默认的 TextinputFormat类读取数据,这个类是由 Hadoop.及其 MapReduce框架提供的。这个输入格式能够读取用换行符换行的文本文件。每读取一行都会调用一次指定的 mapper类的map()方法,这将触发 ImportMapper.map()函数。
ImportMapper类定义了两个方法,这两个方法重载了父类 Mapper,方法名均相同。
ImportMapper类的重载的 setup()方法只会在框架初始化该类时调用一次。在这个例子中,它主要被用于将指定的列信息解析为列族和列限定符。
这个类中的map()才是执行实际工作的实体,它会被输入的文本文件中的每一行调用,每一行都包含了一个JSON格式的记录。这段代码会使用一行数据的MD5散列值来创建一个 HBase行键,然后使用 HBase提供的名为data:json的列存储一行的内容。
这个例子通过 TableoutputFormat类使用了隐式的写缓冲区。调用 context.write()方法时,该方法内部会传入给定的Put实例并调用 table.put()。在作业结束前, TableOutputFormat会主动调用 flushCommits()以保存仍就驻留在写缓冲区的数据。
map()方法可以通过写P吐t实例来存储输入数据,用户也可以通过写 Delete实例来删除目标表中的数据,这就是设置作业输出键的类型为 Writable接口,而并非显式的Put类的原因TableoutputFormat当前仅能处理put与 Delete实例,将消息设置为除Put或 Delete之外的实例都会导致抛出一个 IOException异常。
最后,请注意作业在没有 reduce阶段的情况下,map阶段是怎样工作的。这是相当典型的 HBase与 MapReduce作业结合:由于数据是存储在排序表中的,并且每行数据都拥有唯一的行键,用户可以在流程中避免更消耗的sort、 shuffle和 reduce阶段。
3、数据源
将数据导入到表中之后,我们可以使用其中包含的数据来解析JSON记录,并从中提取信息。这里完全使用了 TableInputFormat类,恰好对应了TableoutputFormat。在 MapReduce处理过程中,它以一张表为输入。
MapReduce作业读取已导入的数据并解析
- 继承已提供的 TableMapper类,设置用户自己的输出键值类型
- 解析JSON数据,提取编辑用户,并统计发生次数。
- 拓展一个 Hadoop Reducer类,并指定适当的类型。
- 统计发生次数,并计算其总和。
- 创建并配置一个scan实例。
- 使用提供的库来设置表的map阶段。
- 使用常规的 Hadoop语法来配置 reduce阶段。
这个作业运行了一个完整的 MapReduce过程,map阶段主要负责从源表中读取JSON数据, reduce阶段主要负责对每个用户做聚合统计。这个例子与 Hadoop提供的 Wordcount例子非常相似, mapper负责记录统计值ONE, reducer负责对每个键做聚合操作,操作
的键是编辑用户(Author)。在命令行上按照如下方式执行作业:
这个例子的结果是包含每个编辑用户的统计列表,并且可以通过命令行方式进行访问,例如, hadoop dfs -text命令:
hadoop dfs -text analyzel/part-r-00000
这个例子展示了怎样使用 TableMapReduceUtil类,通过使用 TableMapReduceUtil的静态方法可以快速配置一个作业,并附带必要的类。由于作业需要 reduce阶段,main()函数代码添加了必要的 Reducer类,并在没有其他特殊指定(本例中是TextoutputFormat类)的情况下隐式地使用默认值。
显然,这是一个非常简单的例子,实际上用户不得不执行更多的分析处理。但即便如此,例子所展现的模式与之前仍类似:用户从表中读取数据,提取必要信息,最终将结果输出到一个特定目标。
4、数据源与数据流向
一个 MapReduce作业中的源和目标都可以是 HBase表,但也有可能在个作业中同时使用 HBase作为输入和输出。换句话说,第三类的 MapReduce模板同时使用HBase表作为输入与输出。这里需要将 TableInputFormat和TableOutputFormat类设置到各自的作业配置项中。
MapReduce作业解析每行数据为若干列
- 将顶层的 JJSON记录的键存储为列,其值的集合存储为列值
- 在 mapper的配置实例中存储列族以备后用。
- 使用通用方法设置map阶段的细节
- 配置一个 reducer实例用于存储解析后的数据。
这个例子使用通用方法配置map阶段和 reduce阶段,并用到了 ParseMapper类,这个类用于从一行原始sON数据中提取信息,还使用了 IdentityTableReducer类将数据存储到目标表。请注意,源表与目标表可以是相同的。用户可以通过以下命令行提交作业:
这个百分比表示map阶段与 reduce阶段已经完成,并且整个作业也已经完成。使用IdentityTableReducer类存储数据并不是必需的,实际上,只需要在代码中增加一行就可以让作业只拥有map阶段。
MapReduce作业解析每行数据为若干列(仅限map阶段)
用户可以从命令行展示的正在运行的作业中看出 reduce阶段已经被跳过了:
reduce阶段一直显示为0%,直到作业完成。用户可以使用 Hadoop MapReduce的UI来确认已执行的作业确实没有 reduce阶段。跳过 reduce阶段的优点是可以更快地完成作业,这是由于框架不需要增加额外的数据处理工作。
这两个不同的 ParseJson作业的作用相同,用户可以通过 HBase Shel看其结果(由于展示空间的缘故,省略重复的行键):
这个导入过程利用了 HBase支持任意列名的特点:JsON的键被转化为列限定符,并形成了新的一列。
5、自定义处理
用户并非必须使用HBae提供的类来读写表数据,实际上这些类非常轻量级并且仅起到了辅助类的作用。改变了前面的代码,以将已解析的JSON数据拆分到两张不同的目标表中,链接和它对应的值存储到了一张名为 linkable的独立的表中,而其他列则存储到了名为 infotable的表中。
MapReduce作业解析每行数据为若干列
- 在setup()方法中创建并配置两张目标表。
- 当任务完成时,刷新所有挂起的提交。
- 保存已解析的数据到两张不同的表中。
- 在mapper的配置实例中存储表名以备后用。
- 设置框架会忽略的输出格式。
用户需要创建两张表,请使用 HBase Shell进行以下操作:
hbase(main): 001: 0> create 'infotable','data'
hbase(main): 002: 0> create 'linktable','data'
在当前例子中,这两张表将会被用作目标存储表通过如下命令行执行作业,并忽略输出内容:
到目前为止,这类似于之前解析JSON的示例,所不同的是结果表及其内容。用户可以在作业完成后使用 HBase Shel和扫描命令罗列内容。用户可以从中看出链接(link)表只包含了链接,而信息(info)表则包含了原JSON数据的剩余字段。
用户可以通过自行编写的 MapReduce代码来控制作业的执行内容。例如,用户可以在存储组合结果的不同表中查询数据。用户从哪里读数据和向哪里写数据是不受限制的。 HBase提供的类是辅助类,或多或少地服务了大量的用例。如果用户发现了它们功能上的限制,可以简单地拓展它们,或直接通过 MapReduce的AP实现可以以任何形式访问 HBase的代码。