Question

Hadoop 应用程序极少由单个 MapReduce Job 或 Hive 脚本构成。分析逻辑通常会分解成几个步骤，再组合成一个执行链来执行完整的分析任务。在之前 MapReduce 与 Apache Web 日志的示例中，我们使用 JobRunners、HiveRunners 和 PigRunners 来执行 HDFS 操作以及 MapReduce、Hive 或 Pig 的 Job，但这并不是一个完全令人满意的解决方案。随着分析链中步骤数量的增加，将会发现执行流程将难以可视化，并且不能在 XML 命名空间中自然地形成图形结构。当我们在重用不同的 Runner 类时，也不能在分析链中跟踪执行步骤。这就意味着在分析链中如果有某个步骤失败了，就必须重新启动（手动）整个分析链，这样使得分析任务的整体执行时间（wall clock）会显著增加并且会影响到效率。本节介绍 Spring Batch 项目的扩展，它提供了将多个 Hadoop 操作链接在一起的构件，通常称为工作流（ workflows ）。

13.2.1 Spring Batch 对 Hadoop 的支持

由于 Hadoop 是面向批处理的系统，所以 Spring Batch 的领域概念和工作流为构造基于 Hadoop 的分析任务奠定了坚实的基础。我们利用组成 Spring Batch Step 的处理动作是可插拔的优势使得 Spring Batch 可被 Hadoop 自动发现。Step 的插入点称为 Tasklet。Spring for Apache Hadoop 提供了自定义的 Tasklet 供 HDFS，以及所有类型的 Hadoop Job：MapReduce、Streaming、Hive 和 Pig 使用。这样就可以创建如图 13-9 所示的工作流。

图 13-9 在 Spring Batch 应用程序中，执行 Hadoop HDFS 操作及运行 Pig、MapReduce 和 Hive Job 的 Step

基于 Eclipse 的 Spring Tool Suite (STS) 支持 Spring Batch Job 可视化编程。图 13-10 展示了在 STS 中与图 13-9 对应的图。

图 13-10 在 Eclipse 中创建 Spring Batch Job

当 Spring Batch Job 的 Step 为线性流程的普通模式时，其底层 XML 配置如示例 13-28 所示。

示例 13-28 按顺序执行 Step 的 Spring Batch Job 定义

也可以根据 Step 的结束状态设置流程条件。在 Step 的返回结果中有一些常见的 ExitStatus 编码，最常见的结束状态是 COMPLETED 和 FAILED。为了创建条件流程，可以使用 Step 的嵌套元素 next，如示例 13-29 所示。

示例 13-29 按条件执行一系列 Step 的 Spring Batch Job 定义

在这个例子中，如果结束代码匹配 FAILED，那下一步要执行的步骤是 stepC，否则在 stepB 之后执行 stepC。

在 Spring 中配置流程的方法很多，本节不再一一讨论。若想要学习更多配置高级工作流的方法，请参见 Spring Batch 参考文档或者之前提到的 Spring Batch 书籍。

可以使用 Spring for Apache Hadoop 提供的 XML 命名空间来配置与 Hadoop 相关的 Step。接下来介绍如何将 wordcount 示例配置成 Spring Batch 应用程序，我们会重用一部分之前 Hadoop 示例中使用过的 MapReduce 配置和 HDFS 脚本；然后介绍如何配置其他与 Hadoop 相关的 Step，例如 Hive 和 Pig。

13.2.2 将 wordcount 样例改造为 Spring Batch 应用

wordcount 示例有两个步骤：将数据导入到 HDFS 中并运行 MapReduce Job。如示例 13-30 所示的 Spring Batch Job 介绍了使用 Spring Batch XML 命名空间的工作流配置。我们使用命名空间前缀 batch 来区别批处理配置和 Hadoop 配置。

示例 13-30 配置 Spring Batch Job 来执行 HDFS 和 MapReduce Step

通常来讲，批处理应用程序的参数化需要 Job 启动时所传入的参数。为了将批处理 Job 改为引用 Job 启动时传入的参数而不是引用静态属性文件，需要对配置进行一些修改。可以使用 SpEL（ http://static.springsource.org/spring/docs/current/spring-framework-reference/ html/expressions.html ）来获取批处理 Job 参数，这和现在所使用的${...}语法引用变量的方式类似。

SpEL 语法和 Java 类似，它的表达式通常只是一行取值代码。我们使用语法#{...}对表达式进行求值来取代使用语法${...}引用变量。为了访问 Spring Batch Job 参数，要使用表达式 #{jobParameters['mr.input']}。在默认情况下，当 Bean 在 Spring Batch 的 Step 作用域内时，变量 jobParameters 是可用的。MapReduce Job 和 HDFS 脚本的配置，如示例 13-31 所示。

示例 13-31 将 Spring Batch Tasklet 链接到 Hadoop Job 以及脚本组件中

运行批处理程序时，应用的主程序会给这些参数传值，也可以使用其他方式为它们赋值以启动 Spring Batch 应用。最常见的选择是通过 CommandLineJobRunner、REST API 管理功能或者 Web 应用管理功能。示例 13-32 为示例应用的主驱动类。

示例 13-32 启动批处理工作主应用程序

为了运行批处理应用，应执行如示例 13-33 所示的命令。

示例 13-33 运行 wordcount 批处理应用脚本

13.2.3 Hive 和 Pig 的步骤

要在 Spring Batch 工作流中执行 Hive 脚本，需使用 Hive Tasklet 元素，如示例 13-34 所示。

示例 13-34 配置 Hive Tasklet

要在 Spring Batch 工作流中执行 Pig 脚本，需使用 Pig Tasklet 元素，如示例 13-35 所示。

示例 13-35 配置 Pig Tasklet