hadoop - Oozie 协调员。如何将过去的数据提供给 mapreduce 作业？

Question

我正在尝试创建 Ooize 协调器。问题是我已经有了等待使用 oozie 处理的暂存数据。

想象一下这样的情况。

1. 当前日期是:01.03.2013(2013 年 3 月 1 日)

2. 我确实有这些输入目录:

   /staging/landing/source/xvlr/2013/02/01/00(2013 年二月一日，一天的第一个小时)/staging/landing/source/xvlr/2013/02/01/01

   /staging/landing/source/xvlr/2013/02/01/02

   /staging/landing/source/xvlr/2013/02/01/03

   /staging/landing/source/xvlr/2013/02/01/04

   ....

   /staging/landing/source/xvlr/2013/02/28/00

   ...

   /staging/landing/source/xvlr/2013/02/28/23

我希望我的 oozie 协调器使用所有以前创建的着陆数据并产生这样的输出:

/masterdata/source/xvlr/2013/02/01/00 
/masterdata/source/xvlr/2013/02/01/01
/masterdata/source/xvlr/2013/02/01/02
/masterdata/source/xvlr/2013/02/01/03
/masterdata/source/xvlr/2013/02/01/04
....
/masterdata/source/xvlr/2013/02/28/00
...
/masterdata/source/xvlr/2013/02/28/23

然后我希望我的协调器每小时运行一次并为主数据生成新的输出。

我如何使用协调器规范来做到这一点？这是我的协调员。它什么都不做。它确实达到了我需要的时间，然后等待。它不会启 Action 业。

请指教。

<coordinator-app name="Xvlr-parser-coordinator" frequency="60"
                 start="2013-03-07T05:35Z" end="2113-01-01T00:35Z" timezone="Europe/Moscow"  xmlns="uri:oozie:coordinator:0.3">
    <controls>
        <timeout>5</timeout>
        <concurrency>4</concurrency>
    </controls>

    <datasets>
        <dataset name="xvlrInputDataset" frequency="${coord:hours(1)}" initial-instance="2013-03-07T05:35Z" timezone="Europe/Moscow">
            <uri-template>${nameNode}/staging/landing/source/xvlr/${YEAR}/${MONTH}/${DAY}/${HOUR}</uri-template>
            <done-flag></done-flag>
        </dataset>
        <dataset name="xvlrOutputDataset" frequency="${coord:hours(1)}" initial-instance="2013-03-07T05:35Z" timezone="Europe/Moscow">
            <uri-template>${nameNode}/masterdata/source/xvlr/archive/${YEAR}/${MONTH}/${DAY}/${HOUR}</uri-template>
            <done-flag></done-flag>
        </dataset>

    </datasets>

    <input-events>
        <data-in name="xvlrInputEvent" dataset="xvlrInputDataset">
            <instance>${coord:current(0)}</instance>
        </data-in>
    </input-events>

    <output-events>
        <data-out name="xvlrOutputEvent" dataset="xvlrOutputDataset">
            <instance>${coord:current(0)}</instance>
        </data-out>
    </output-events>
    <action>
        <workflow>
            <app-path>${oozieAppHomeCatalog}/sub-workflows/Xvlr-parser-subworkflow.xml</app-path>
            <configuration>
                <property>
                    <name>inputDir</name>
                    <value>${coord:dataIn('xvlrInputEvent')}</value>
                </property>
                <property>
                    <name>outputDir</name>
                    <value>${coord:dataOut('xvlrOutputEvent')}</value>
                </property>

            </configuration>

        </workflow>
    </action>
</coordinator-app>

Answer 1

这是正确的解决方案(它可以工作几天:))):

<coordinator-app name="Xvlr-parser-coordinator" frequency="60"
                 start="2013-03-07T16:35Z" end="2113-01-01T00:35Z" timezone="Europe/Moscow"  xmlns="uri:oozie:coordinator:0.3">
    <controls>
        <timeout>3</timeout>
        <concurrency>1</concurrency>
    </controls>

    <datasets>
        <dataset name="xvlrInputDataset" frequency="${coord:hours(1)}" initial-instance="2013-03-07T16:35Z" timezone="Europe/Moscow">
            <uri-template>${nameNode}/staging/landing/source/xvlr/${YEAR}/${MONTH}/${DAY}/${HOUR}</uri-template>
            <done-flag></done-flag>
        </dataset>
        <dataset name="xvlrOutputDataset" frequency="${coord:hours(1)}" initial-instance="2013-03-07T16:35Z" timezone="Europe/Moscow">
            <uri-template>${nameNode}/masterdata/source/xvlr/archive/${YEAR}/${MONTH}/${DAY}/${HOUR}</uri-template>
            <done-flag></done-flag>
        </dataset>

    </datasets>

    <input-events>
        <data-in name="xvlrInputEvent" dataset="xvlrInputDataset">
            <instance>${coord:current(0)}</instance>
        </data-in>
    </input-events>

    <output-events>
        <data-out name="xvlrOutputEvent" dataset="xvlrOutputDataset">
            <instance>${coord:current(0)}</instance>
        </data-out>
    </output-events>
    <action>
        <workflow>
            <app-path>${oozieAppHomeCatalog}/sub-workflows/Xvlr-parser-subworkflow.xml</app-path>
            <configuration>
                <property>
                    <name>inputDir</name>
                    <value>${coord:dataIn('xvlrInputEvent')}</value>
                </property>
                <property>
                    <name>outputDir</name>
                    <value>${coord:dataOut('xvlrOutputEvent')}</value>
                </property>
            </configuration>

        </workflow>
    </action>
</coordinator-app>

它有什么作用？

• 起初它确实是从 2013-03-07T16:35Z 开始的，所以之前的所有内容
  收集的数据已通过底层工作流(一个 mr-job具有解析功能的调用)
  • 在处理“过去时间数据集”(数据集时间小于当前时间)时，工作流正在一个接一个地运行:它确实消耗了/pastdate/hour_00，然后它立即开始消耗/pastdate/hour_01，等等
  • 当协调器到达当前时间时，它开始每小时调用一次工作流(按设计:05:35、06:35 ... 23:35)。
  • 查看超时声明:我确实缺少数据集:例如，3 月 1 日的第 10 个小时没有数据。工作流确实等待了 3 分钟，然后就死了。

问题解决了。