第 9 章? 定期自动执行ETL作业 ?

一旦数据仓库开始使用，就需要不断从源系统给数据仓库提供新数据。为了确保数据流的稳定，需要使用所在平台上可用的任务调度器来调度ETL定期执行。调度模块是ETL系统必不可少的组成部分，它不但是数据仓库的基本需求，也对项目的成功起着举足轻重的作用。操作系统一般都为用户提供调度作业的功能，如Windows的“计划任务”和UNIX/Linux的cron系统服务。绝大多数Hadoop系统都运行在Linux之上，因此本章详细讨论两种Linux上定时自动执行ETL作业的方案。一种是经典的crontab，这是操作系统自带的功能，二是Hadoop生态圈中的Oozie组件。为了演示Hadoop对数据仓库的支持能力，我们的示例将使用后者实现ETL执行自动化。9.1 crontab上一章我们已经准备好用于定期装载的regular_etl.sh shell脚本文件，可以很容易地用crontab命令创建一个任务，定期运行此脚本。# 修改文件属性为可执行chmod 755 /root/regular_etl.sh# 编辑crontab文件内容crontab -e# 添加如下一行，指定每天2点执行定期装载作业，然后保存退出0 2 /root/regular_etl.sh这就可以了，需要用户做的就是如此简单，其他的事情交给cron系统服务去完成。提供cron服务的进程名为crond，这是Linux下一个用来周期性执行某种任务或处理某些事件的守护进程。当安装完操作系统后，会自动启动crond进程，它每分钟会定期检查是否有要执行的任务，如果有则自动执行该任务。Linux下的任务调度分为两类，系统任务调度和用户任务调度。? 系统任务调度：系统需要周期性执行的工作，比如写缓存数据到硬盘、日志清理等。在/etc目录下有一个crontab文件，这个就是系统任务调度的配置文件。? 用户任务调度：用户要定期执行的工作，比如用户数据备份、定时邮件提醒等。用户可以使用crontab命令来定制自己的计划任务。所有用户定义的crontab文件都被保存在 /var/spool/cron目录中，其文件名与用户名一致。1. crontab权限Linux系统使用一对allow/deny文件组合判断用户是否具有执行crontab的权限。如果用户名出现在/etc/cron.allow文件中，则该用户允许执行crontab命令。如果此文件不存在，那么如果用户名没有出现在/etc/cron.deny文件中，则该用户允许执行crontab命令。如果只存在cron.deny文件，并且该文件是空的，则所有用户都可以使用crontab命令。如果这两个文件都不存在，那么只有root用户可以执行crontab命令。allow/deny文件由每行一个用户名构成。2. crontab命令通过crontab 命令，我们可以在固定间隔的时间点执行指定的系统指令或 shell脚本。时间间隔的单位可以是分钟、小时、日、月、周及以上的任意组合。crontab 命令格式如下：crontab [-u user] filecrontab [-u user] [ -e | -l | -r ]说明：? -u user：用来设定某个用户的crontab服务，此参数一般由root用户使用。? file：file是命令文件的名字，表示将file作为crontab的任务列表文件并载入crontab。如果在命令行中没有指定这个文件，crontab命令将接受标准输入，通常是键盘上键入的命令，并将它们载入crontab。? -e：编辑某个用户的crontab文件内容。如果不指定用户，则表示编辑当前用户的crontab文件。如果文件不存在，则创建一个。? -l：显示某个用户的crontab文件内容，如果不指定用户，则表示显示当前用户的crontab文件内容。? -r：从/var/spool/cron目录中删除某个用户的crontab文件，如果不指定用户，则默认删除当前用户的crontab文件。注意：如果不经意地输入了不带任何参数的crontab命令，不要使用Control-d退出，因为这会删除用户所对应的crontab文件中的所有条目。代替的方法是用Control-c退出。3. crontab文件用户所建立的crontab文件中，每一行都代表一项任务，每行的每个字段代表一项设置。它的格式共分为六个字段，前五段是时间设定段，第六段是要执行的命令段，格式如下：.---------------- 分钟(0 - 59)| .------------- 小时(0 - 23)| | .---------- 日期(1 - 31)| | | .------- 月份(1 - 12)| | | | .---- 星期(0 - 6，代表周日到周一) | | | | | 要执行的命令，可以是系统命令，也可以是自己编写的脚本文件。在以上各个时间字段中，还可以使用如下特殊字符：? 星号()：代表所有可能的值，例如“月份”字段如果是星号，则表示在满足其他字段的制约条件后每月都执行该命令操作。? 逗号(,)：可以用逗号隔开的值指定一个列表范围，例如，“1,2,5,7,8,9”。? 中杠(-)：可以用整数之间的中杠表示一个整数范围，例如“2-6”表示“2,3,4,5,6”。? 正斜线(/)：可以用正斜线指定时间的间隔频率，例如“0-23/2”表示每两小时执行一次。同时正斜线可以和星号一起使用，例如/10，如果用在“分钟”字段，表示每十分钟执行一次。注意，“日期”和“星期”字段都可以指定哪天执行，如果两个字段都设置了，则执行的日期是两个字段的并集。4. crontab示例# 每1分钟执行一次command command# 每小时的第3和第15分钟执行3,15 command# 在上午8点到11点的第3和第15分钟执行3,15 8-11 command# 每隔两天的上午8点到11点的第3和第15分钟执行3,15 8-11 /2 command# 每个星期一的上午8点到11点的第3和第15分钟执行3,15 8-11 1 command# 每晚的21:30执行30 21 command# 每月1、10、22日的4:45执行 45 4 1,10,22 command# 每周六、周日的1:10执行10 1 6,0 command# 每天18:00至23:00之间每隔30分钟执行 0,30 18-23 command# 每星期六的晚上11:00执行 0 23 6 command# 每一小时执行一次 /1 command# 晚上11点到早上7点之间，每隔一小时执行一次 23-7/1 command# 每月的4号与每周一到周三的11点执行 0 11 4 1-3 command# 一月一号的4点执行 0 4 1 1 command# 每小时执行/etc/cron.hourly目录内的脚本01 root run-parts /etc/cron.hourly说明：run-parts会遍历目标文件夹，执行及时层目录下具有可执行权限的文件。5. crontab环境有时我们创建了一个crontab任务，但是这个任务却无法自动执行，而手动执行脚本却没有问题，这种情况一般是由于在crontab文件中没有配置环境变量引起的。cron从用户所在的主目录中使用shell调用需要执行的命令。cron为每个shell提供了一个默认的环境，Linux下的定义如下：SHELL=/bin/bashPATH=/sbin:/bin:/usr/sbin:/usr/binMAILTO=用户名HOME=用户主目录在crontab文件中定义多个调度任务时，需要特别注意的一个问题就是环境变量的设置，因为我们手动执行某个脚本时，是在当前shell环境下进行的，程序能找到环境变量；而系统自动执行任务调度时，除了默认的环境，是不会加载任何其他环境变量的。因此就需要在crontab文件中指定任务运行所需的所有环境变量。不要假定cron知道所需要的特殊环境，它其实并不知道。所以用户要保障在shell脚本中提供所有必要的路径和环境变量，除了一些自动设置的全局变量。以下三点需要注意：? 脚本中涉及文件路径时写路径；? 脚本执行要用到环境变量时，通过source命令显式引入，例如：#!/bin/shsource /etc/profile? 当手动执行脚本没问题，但是crontab不执行时，可以尝试在crontab中直接引入环境变量解决问题，例如：0 . /etc/profile;/bin/sh /path/to/myscript.sh6. 重定向输出邮件默认时，每条任务调度执行完毕，系统都会将任务输出信息通过电子邮件的形式发送给当前系统用户。这样日积月累，日志信息会非常大，可能会影响系统的正常运行。因此，将每条任务进行重定向处理非常重要。可以在crontab文件中设置如下形式，忽略日志输出：0 /3 /usr/local/myscript.sh >/dev/null 2>&1“>/dev/null 2>&1”表示先将标准输出重定向到/dev/null，然后将标准错误重定向到标准输出。由于标准输出已经重定向到了/dev/null，因此标准错误也会重定向到/dev/null，这样日志输出问题就解决了。7. 生成日志文件可以将crontab执行任务的输出信息重定向到一个自定义的日志文件中，例如：8 rm /home/someuser/tmp/ > /home/someuser/cronlogs/clean_tmp_dir.log9.2 Oozie简介除了利用操作系统提供的功能以外，Hadoop生态圈的工具也可以完成同样的调度任务，而且更灵活，这个组件就是Oozie。Oozie是一个管理Hadoop作业、可伸缩、可扩展、的工作流调度系统，它内部定义了三种作业：工作流作业、协调器作业和Bundle作业。工作流作业是由一系列动作构成的有向无环图(DAGs)，协调器作业是按时间频率周期性触发Oozie工作流的作业，Bundle管理协调器作业。Oozie支持的用户作业类型有Java map-reduce、Streaming map-reduce、Pig、 Hive、Sqoop和Distcp，及其Java程序和shell脚本或命令等特定的系统作业。Oozie项目经历了三个主要阶段。及时版Oozie是一个基于工作流引擎的服务器，通过执行Hadoop MapReduce和Pig作业的动作运行工作流作业。第二版Oozie是一个基于协调器引擎的服务器，按时间和数据触发工作流执行。它可以基于时间(如每小时执行一次)或数据可用性(如等待输入数据完成后再执行)连续运行工作流。第三版Oozie是一个基于Bundle引擎的服务器。它提供更高级别的抽象，批量处理一系列协调器应用。用户可以在bundle级别启动、停止、挂起、继续、重做协调器作业，这样可以更好地简化操作控制。使用Oozie主要基于以下两点原因：? 在Hadoop中执行的任务有时候需要把多个MapReduce作业连接到一起执行，或者需要多个作业并行处理。Oozie可以把多个MapReduce作业组合到一个逻辑工作单元中，从而完成更大型的任务。? 从调度的角度看，如果使用crontab的方式调用多个工作流作业，可能需要编写大量的脚本，还要通过脚本来控制好各个工作流作业的执行时序问题，不但不好维护，而且监控也不方便。基于这样的背景，Oozie提出了Coordinator的概念，它能够将每个工作流作业作为一个动作来运行，相当于工作流定义中的一个执行节点，这样就能够将多个工作流作业组成一个称为Coordinator Job的作业，并指定触发时间和频率，还可以配置数据集、并发数等。9.2.1 Oozie的体系结构Oozie的体系结构如图9-1所示。 图9-1 Oozie体系结构Oozie是一种Java Web应用程序，它运行在Java Servlet容器，即Tomcat中，并使用数据库来存储以下内容：? 工作流定义。? 当前运行的工作流实例，包括实例的状态和变量。Oozie工作流是放置在DAG(有向无环图 Direct Acyclic Graph)中的一组动作，例如，Hadoop的Map/Reduce作业、Pig作业等。DAG控制动作的依赖关系，指定了动作执行的顺序。Oozie使用hPDL这种XML流程定义语言来描述这个图。hPDL是一种很简洁的语言，它只会使用少数流程控制节点和动作节点。控制节点会定义执行的流程，并包含工作流的起点和终点(start、end和fail节点)以及控制工作流执行路径的机制(decision、fork和join节点)。动作节点是实际执行操作的部分，通过它们工作流会触发执行计算或者处理任务。Oozie为以下类型的动作提供支持：Hadoop MapReduce、Hadoop HDFS、Pig、Java和Oozie的子工作流。而SSH动作已经从Oozie schema 0.2之后的版本中移除了。所有由动作节点触发的计算和处理任务都不在Oozie中运行。它们是由Hadoop的MapReduce框架执行的。这种低耦合的设计方法让Oozie可以有效利用Hadoop的负载平衡、灾难恢复等机制。这些任务主要是串行执行的，只有文件系统动作例外，它是并行处理的。这意味着对于大多数工作流动作触发的计算或处理任务类型来说，在工作流操作转换到工作流的下一个节点之前都需要等待，直到前面节点的计算或处理任务结束了之后才能够继续。Oozie可以通过两种不同的方式来检测计算或处理任务是否完成，这就是回调和轮询。当Oozie启动了计算或处理任务时，它会为任务提供的回调URL，然后任务会在完成的时候发送通知给这个特定的URL。在任务无法触发回调URL的情况下(可能是因为任何原因，比方说网络闪断)，或者当任务的类型无法在完成时触发回调URL的时候，Oozie有一种机制，可以对计算或处理任务进行轮询，从而能够判断任务是否完成。Oozie工作流可以参数化，例如在工作流定义中使用像${inputDir}之类的变量等。在提交工作流操作的时候，我们必须提供参数值。如果经过合适地参数化，比如使用不同的输出目录，那么多个同样的工作流操作可以并发执行。一些工作流是根据需要触发的，但是大多数情况下，我们有必要基于一定的时间段、数据可用性或外部事件来运行它们。Oozie协调系统(Coordinator system)让用户可以基于这些参数来定义工作流执行计划。Oozie协调程序让我们可以用谓词的方式对工作流执行触发器进行建模，谓词可以是时间条件、数据条件、内部事件或外部事件。工作流作业会在谓词得到满足的时候启动。不难看出，这里的谓词，其作用和SQL语句的WHERE子句中的谓词类似，本质上都是在满足某些条件时触发某种事件。有时，我们还需要连接定时运行、但时间间隔不同的工作流操作。多个以不同频率运行的工作流的输出会成为下一个工作流的输入。把这些工作流连接在一起，会让系统把它作为数据应用的管道来引用。Oozie协调程序支持创建这样的数据应用管道。9.2.2 CDH 5.7.0中的OozieCDH 5.7.0中，Oozie的版本是4.1.0，其元数据存储使用MySQL(4.4节CDH安装中有相关配置)。关于CDH 5.7.0中Oozie的属性，参考以下链接：https://www.cloudera.com/documentation/enterprise/latest/topics/cm_props_cdh570_oozie.html9.3 建立定期装载工作流对于刚接触Oozie的用户来说，前面介绍的概念过于抽象，不易理解，那么就让我们一步步创建销售订单示例ETL的工作流，在实例中学习Oozie的特性和用法。1. 修改资源配置Oozie运行需要使用较高的内存资源，因此要将以下两个YARN参数的值调大：? yarn.nodemanager.resource.memory-mb：NodeManage总的可用物理内存。? yarn.scheduler.maximum-allocation-mb：一个MapReduce任务可申请的较大内存。如果分配的内存不足，在执行工作流作业时会报类似下面的错误：org.apache.oozie.action.ActionExecutorException: JA009: org.apache.hadoop.yarn.exceptions.InvalidResourceRequestException: Invalid resource request, requested memory < 0, or requested memory > max configured, requestedMemory=1536, maxMemory=1500我们的实验环境中，每个Hadoop节点所在虚拟机的总物理内存为8GB，所以把这两个参数都设置为2GB。修改的方法有两种，可以编辑yarn-site.xml文件里的属性，如：yarn.nodemanager.resource.memory-mb 2000 yarn.scheduler.maximum-allocation-mb 2000或者在Cloudera Manager中修改，yarn.nodemanager.resource.memory-mb参数在YARN服务的NodeManager范围里，yarn.scheduler.maximum-allocation-mb参数在YARN服务的ResourceManager范围里。无论使用哪种方法，修改后都需要保存更改并重启Hadoop集群。2. 启用Oozie Web Console默认安装CDH时，Oozie Web Console是禁用的，为了后面方便监控Oozie作业的执行，需要将其改为启用状态。“启用 Oozie 服务器 Web 控制台”属性在Oozie服务的“Oozie Server Default Group”里。具体的做法是： 下载ext-2.2包，解压缩到Oozie服务器实例所在节点的/var/lib/oozie/目录下。 登录Cloudera Manager管理控制台，进入Oozie服务页面。 单击“配置”标签。 定位“启用 Oozie 服务器 Web 控制台”属性，或者在搜索框中输入该属性名查找。 选择“启用 Oozie 服务器 Web 控制台”的复选框。 单击“保存更改”按钮提交所做的修改。 重启Oozie服务。3. 启动Sqoop的share metastore service定期装载工作流需要用Oozie调用Sqoop执行，这需要开启Sqoop的元数据共享存储，命令如下：sqoop metastore > /tmp/sqoop_metastore.log 2>&1 &metastore工具配置Sqoop作业的共享元数据信息存储，它会在当前主机启动一个内置的HSQLDB共享数据库实例。客户端可以连接这个metastore，这样允许多个用户定义并执行metastore中存储的Sqoop作业。metastore库文件的存储位置由sqoop-site.xml中的sqoop.metastore.server.location属性配置，它指向一个本地文件。如果不设置这个属性，Sqoop元数据默认存储在~/.sqoop目录下。如果碰到用Oozie工作流执行Sqoop命令是成功的，但执行Sqoop作业却失败的情况，可以参考“Oozie系列(3)之解决Sqoop Job无法运行的问题”这篇文章。该文中对这个问题有很详细的分析，并提供了解决方案，其访问地址是：www.lamborryan.com/oozie-sqoop-fail/4. 连接metastore重建sqoop job我们在上一章中建立了一个增量抽取sales_order表数据的Sqoop作业，但其元数据并没有存储在shared metastore里，所以需要使用以下的命令进行重建：last_value=`sqoop job --show myjob_incremental_import | grep incremental.last.value | awk '{print $3}'` sqoop job --delete myjob_incremental_import sqoop job \ --meta-connect jdbc:hsqldb:hsql://cdh2:16000/sqoop \ --create myjob_incremental_import \ -- \ import \ --connect "jdbc:mysql://cdh1:3306/source?useSSL=false&user=root&password=mypassword" \ --table sales_order \ --columns "order_number, customer_number, product_code, order_date, entry_date, order_amount" \ --hive-import \ --hive-table rds.sales_order \ --incremental append \ --check-column order_number \ --last-value $last_value 在上面命令的及时行中，先用sqoop的--show选项查询一次执行定期装载后incremental.last.value的值，并将这个值赋给名为last_value的shell变量。在创建作业命令的一行中，--last-value选项将当前较大值作为参数。和上一章的myjob_incremental_import作业创建命令对比，会发现多了一行--meta-connect jdbc:hsqldb:hsql://cdh2:16000/sqoop，就是通过这个选项将作业元数据存储到HSQLDB数据库文件中，metastore的默认端口是16000，可以用sqoop.metastore.server.port属性设置为其他端口号。创建作业前，需要使用--delete参数先删除已经存在的同名作业。5. 定义工作流建立内容如下的workflow.xml文件： &