4.1.1 End of the day in Datetime Values

It is valid to express the end of day as a '24:00' hour (i.e. 2009-08-10T24:00Z ).

However, for all calculations and display, Oozie resolves such dates as the zero hour of the following day (i.e. 2009-08-11T00:00Z ).

4.4.1. The coord:days(int n) and coord:endOfDays(int n) EL functions

The ${coord:days(int n)} and ${coord:endOfDays(int n)} EL functions should be used to handle day based frequencies.

Constant values should not be used to indicate a day based frequency (every 1 day, every 1 week, etc) because the number of hours in every day is not always the same for timezones that observe daylight-saving time.

It is a good practice to use always these EL functions instead of using a constant expression (i.e. 24 * 60 ) even if the timezone for which the application is being written for does not support daylight saving time. This makes application foolproof to country legislations changes and also makes applications portable across timezones.

<coordinator-app name="hello-coord" frequency="${coord:days(1)}"
                  start="2009-01-02T08:00Z" end="2009-01-04T08:00Z" timezone="America/Los_Angeles"
                 xmlns="uri:oozie:coordinator:0.1">
      <controls>
        <timeout>10</timeout>
        <concurrency>${concurrency_level}</concurrency>
        <execution>${execution_order}</execution>
        <throttle>${materialization_throttle}</throttle>
      </controls>      <datasets>
       <dataset name="din" frequency="${coord:endOfDays(1)}"
                initial-instance="2009-01-02T08:00Z" timezone="America/Los_Angeles">
         <uri-template>${baseFsURI}/${YEAR}/${MONTH}/${DAY}/${HOUR}/${MINUTE}</uri-template>
        </dataset>
       <dataset name="dout" frequency="${coord:minutes(30)}"
                initial-instance="2009-01-02T08:00Z" timezone="UTC">
         <uri-template>${baseFsURI}/${YEAR}/${MONTH}/${DAY}/${HOUR}/${MINUTE}</uri-template>
        </dataset>
      </datasets>
      <input-events>
         <data-in name="input" dataset="din">
				<instance>${coord:current(0)}</instance>
         </data-in>
      </input-events>
      <output-events>
         <data-out name="output" dataset="dout">
				<instance>${coord:current(1)}</instance>
         </data-out>
      </output-events>
      <action>
        <workflow>
          <app-path>${wf_app_path}</app-path>
          <configuration>
              <property>
              <name>wfInput</name>
              <value>${coord:dataIn('input')}</value>
            </property>
            <property>
              <name>wfOutput</name>
              <value>${coord:dataOut('output')}</value>
            </property>
         </configuration>
       </workflow>
      </action>
 </coordinator-app>

4.4.2. The coord:months(int n) and coord:endOfMonths(int n) EL functions

The ${coord:months(int n)} and ${coord:endOfMonths(int n)} EL functions should be used to handle month based frequencies.

Constant values cannot be used to indicate a month based frequency because the number of days in a month changes month to month and on leap years; plus the number of hours in every day of the month are not always the same for timezones that observe daylight-saving time.

4.4.2.1. The coord:months(int n) EL function

The ${coord:months(int n)} EL function returns the number of minutes for 'n' complete months starting with the month of the current nominal time for which the computation is being done.

The ${coord:months(int n)} EL function includes all the minutes of the current month, regardless of the day of the month of the current nominal time.

Examples:

4.4.2.2. The coord:endOfMonths(int n) EL function

The ${coord:endOfMonths(int n)} EL function is identical to the ${coord:months(int n)} except that it shifts the first occurrence to the end of the month for the specified timezone before computing the interval in minutes.

Examples:

<coordinator-app name="hello-coord" frequency="${coord:months(1)}"
                  start="2009-01-02T08:00Z" end="2009-04-02T08:00Z" timezone="America/Los_Angeles"
                 xmlns="uri:oozie:coordinator:0.1">
      <controls>
        <timeout>10</timeout>
        <concurrency>${concurrency_level}</concurrency>
        <execution>${execution_order}</execution>
        <throttle>${materialization_throttle}</throttle>
      </controls>      <datasets>
       <dataset name="din" frequency="${coord:endOfMonths(1)}"
                initial-instance="2009-01-02T08:00Z" timezone="America/Los_Angeles">
         <uri-template>${baseFsURI}/${YEAR}/${MONTH}/${DAY}/${HOUR}/${MINUTE}</uri-template>
        </dataset>
       <dataset name="dout" frequency="${coord:minutes(30)}"
                initial-instance="2009-01-02T08:00Z" timezone="UTC">
         <uri-template>${baseFsURI}/${YEAR}/${MONTH}/${DAY}/${HOUR}/${MINUTE}</uri-template>
        </dataset>
      </datasets>
      <input-events>
         <data-in name="input" dataset="din">
				<instance>${coord:current(0)}</instance>
         </data-in>
      </input-events>
      <output-events>
         <data-out name="output" dataset="dout">
				<instance>${coord:current(1)}</instance>
         </data-out>
      </output-events>
      <action>
        <workflow>
          <app-path>${wf_app_path}</app-path>
          <configuration>
              <property>
              <name>wfInput</name>
              <value>${coord:dataIn('input')}</value>
            </property>
            <property>
              <name>wfOutput</name>
              <value>${coord:dataOut('output')}</value>
            </property>
         </configuration>
       </workflow>
      </action>
 </coordinator-app>

6.1.1. Coordinator Application

A coordinator application is a program that triggers actions (commonly workflow jobs) when a set of conditions are met. Conditions can be a time frequency, the availability of new dataset instances or other external events.

Types of coordinator applications:

• Synchronous: Its coordinator actions are created at specified time intervals.

Coordinator applications are normally parameterized.

6.1.2. Coordinator Job

To create a coordinator job, a job configuration that resolves all coordinator application parameters must be provided to the coordinator engine.

A coordinator job is a running instance of a coordinator application running from a start time to an end time.

At any time, a coordinator job is in one of the following status: PREP, RUNNING, RUNNINGWITHERROR, PREPSUSPENDED, SUSPENDED, SUSPENDEDWITHERROR, PREPPAUSED, PAUSED, PAUSEDWITHERROR, SUCCEEDED, DONEWITHERROR, KILLED, FAILED .

Valid coordinator job status transitions are:

• PREP --> PREPSUSPENDED | PREPPAUSED | RUNNING | KILLED
• RUNNING --> RUNNINGWITHERROR | SUSPENDED | PAUSED | SUCCEEDED | KILLED
• RUNNINGWITHERROR --> RUNNING | SUSPENDEDWITHERROR | PAUSEDWITHERROR | DONEWITHERROR | KILLED | FAILED
• PREPSUSPENDED --> PREP | KILLED
• SUSPENDED --> RUNNING | KILLED
• SUSPENDEDWITHERROR --> RUNNINGWITHERROR | KILLED
• PREPPAUSED --> PREP | KILLED
• PAUSED --> SUSPENDED | RUNNING | KILLED
• PAUSEDWITHERROR --> SUSPENDEDWITHERROR | RUNNINGWITHERROR | KILLED

When a coordinator job is submitted, oozie parses the coordinator job XML. Oozie then creates a record for the coordinator with status PREP and returns a unique ID. The coordinator is also started immediately if pause time is not set.

When a user requests to suspend a coordinator job that is in PREP state, oozie puts the job in status PREPSUSPENDED . Similarly, when pause time reaches for a coordinator job with PREP status, oozie puts the job in status PREPPAUSED .

Conversely, when a user requests to resume a PREPSUSPENDED coordinator job, oozie puts the job in status PREP . And when pause time is reset for a coordinator job and job status is PREPPAUSED , oozie puts the job in status PREP .

When a coordinator job starts, oozie puts the job in status RUNNING and start materializing workflow jobs based on job frequency. If any workflow job goes to FAILED/KILLED/TIMEDOUT state, the coordinator job is put in RUNNINGWITHERROR

When a user requests to kill a coordinator job, oozie puts the job in status KILLED and it sends kill to all submitted workflow jobs.

When a user requests to suspend a coordinator job that is in RUNNING status, oozie puts the job in status SUSPENDED and it suspends all submitted workflow jobs. Similarly, when a user requests to suspend a coordinator job that is in RUNNINGWITHERROR status, oozie puts the job in status SUSPENDEDWITHERROR and it suspends all submitted workflow jobs.

When pause time reaches for a coordinator job that is in RUNNING status, oozie puts the job in status PAUSED . Similarly, when pause time reaches for a coordinator job that is in RUNNINGWITHERROR status, oozie puts the job in status PAUSEDWITHERROR .

Conversely, when a user requests to resume a SUSPENDED coordinator job, oozie puts the job in status RUNNING . Also, when a user requests to resume a SUSPENDEDWITHERROR coordinator job, oozie puts the job in status RUNNINGWITHERROR . And when pause time is reset for a coordinator job and job status is PAUSED , oozie puts the job in status RUNNING . Also, when the pause time is reset for a coordinator job and job status is PAUSEDWITHERROR , oozie puts the job in status RUNNINGWITHERROR

A coordinator job creates workflow jobs (commonly coordinator actions) only for the duration of the coordinator job and only if the coordinator job is in RUNNING status. If the coordinator job has been suspended, when resumed it will create all the coordinator actions that should have been created during the time it was suspended, actions will not be lost, they will delayed.

When the coordinator job materialization finishs and all workflow jobs finish, oozie updates the coordinator status accordingly. For example, if all workflows are SUCCEEDED , oozie puts the coordinator job into SUCCEEDED status. If all workflows are FAILED , oozie puts the coordinator job into FAILED status. If all workflows are KILLED , the coordinator job status changes to KILLED. However, if any workflow job finishes with not SUCCEEDED and combination of KILLED , FAILED or TIMEOUT , oozie puts the coordinator job into DONEWITHERROR . If all coordinator actions are TIMEDOUT , oozie puts the coordinator job into DONEWITHERROR .

6.1.3. Coordinator Action

A coordinator job creates and executes coordinator actions.

A coordinator action is normally a workflow job that consumes and produces dataset instances.

Once an coordinator action is created (this is also referred as the action being materialized), the coordinator action will be in waiting until all required inputs for execution are satisfied or until the waiting times out.

6.1.4. Input Events

The Input events of a coordinator application specify the input conditions that are required in order to execute a coordinator action.

In the current specification input events are restricted to dataset instances availability.

All the datasets instances defined as input events must be available for the coordinator action to be ready for execution ( READY status).

Input events are normally parameterized. For example, the last 24 hourly instances of the 'searchlogs' dataset.

Input events can be refer to multiple instances of multiple datasets. For example, the last 24 hourly instances of the 'searchlogs' datset and the last weekly instance of the 'celebrityRumours' dataset.

6.1.5. Output Events

A coordinator action can produce one or more dataset(s) instances as output.

Dataset instances produced as output by one coordinator actions may be consumed as input by another coordinator action(s) of other coordinator job(s).

The chaining of coordinator jobs via the datasets they produce and consume is referred as a data pipeline.

In the current specification coordinator job output events are restricted to dataset instances.

6.1.6. Coordinator Action Execution Policies

The execution policies for the actions of a coordinator job can be defined in the coordinator application.

• Timeout: A coordinator job can specify the timeout for its coordinator actions, this is, how long the coordinator action will be in WAITING or READY status before giving up on its execution.
• Concurrency: A coordinator job can specify the concurrency for its coordinator actions, this is, how many coordinator actions are allowed to run concurrently ( RUNNING status) before the coordinator engine starts throttling them.
• Execution strategy: A coordinator job can specify the execution strategy of its coordinator actions when there is backlog of coordinator actions in the coordinator engine. The different execution strategies are 'oldest first', 'newest first' and 'last one only'. A backlog normally happens because of delayed input data, concurrency control or because manual re-runs of coordinator jobs.
• Throttle: A coordinator job can specify the materialization or creation throttle value for its coordinator actions, this is, how many maximum coordinator actions are allowed to be in WAITING state concurrently.

6.1.7. Data Pipeline Application

Commonly, multiple workflow applications are chained together to form a more complex application.

Workflow applications are run on regular basis, each of one of them at their own frequency. The data consumed and produced by these workflow applications is relative to the nominal time of workflow job that is processing the data. This is a coordinator application .

The output of multiple workflow jobs of a single workflow application is then consumed by a single workflow job of another workflow application, this is done on regular basis as well. These workflow jobs are triggered by recurrent actions of coordinator jobs. This is a set of coordinator jobs that inter-depend on each other via the data they produce and consume.

This set of interdependent coordinator applications is referred as a data pipeline application .

6.6.1. coord:current(int n) EL Function for Synchronous Datasets

${coord:current(int n)} represents the n^th dataset instance for a synchronous dataset, relative to the coordinator action creation (materialization) time. The coordinator action creation (materialization) time is computed based on the coordinator job start time and its frequency. The n^th dataset instance is computed based on the dataset's initial-instance datetime, its frequency and the (current) coordinator action creation (materialization) time.

n can be a negative integer, zero or a positive integer.

${coord:current(int n)} returns the nominal datetime for n^th dataset instance relative to the coordinator action creation (materialization) time.

${coord:current(int n)} performs the following calculation:

DS_II : dataset initial-instance (datetime)
DS_FREQ: dataset frequency (minutes)
CA_NT: coordinator action creation (materialization) nominal timecoord:current(int n) = DS_II + DS_FREQ * ( (CA_NT - DS_II) div DS_FREQ + n)

NOTE: The formula above is not 100% correct, because DST changes the calculation has to account for hour shifts. Oozie Coordinator must make the correct calculation accounting for DTS hour shifts.

When a positive integer is used with the ${coord:current(int n)} , it refers to a dataset instance in the future from the coordinator action creation (materialization) time. This can be useful when creating dataset instances for future use by other systems.

The datetime returned by ${coord:current(int n)} returns the exact datetime for the computed dataset instance.

IMPORTANT: The coordinator engine does use output events to keep track of new dataset instances. Workflow jobs triggered from coordinator actions can leverage the coordinator engine capability to synthesize dataset instances URIs to create output directories.

Examples:

1. *=${coord:current(int n)}= datetime calculation:*

Datasets Definition:

<datasets>
.
  <dataset name="logs" frequency="${coord:days(1)}"
           initial-instance="2009-01-01T24:00Z" timezone="UTC">
    <uri-template>hdfs://bar:8020/app/logs/${YEAR}${MONTH}/${DAY}</uri-template>
  </dataset>
.
  <dataset name="weeklySiteAccessStats" frequency="${coord:days(7)}"
           initial-instance="2009-01-07T24:00Z" timezone="UTC">
    <uri-template>hdfs://bar:8020/app/weeklystats/${YEAR}/${MONTH}/${DAY}</uri-template>
  </dataset>
.
</datasets>

For a coordinator action creation time: 2009-05-29T24:00Z the ${coord:current(int n)} EL function would resolve to the following datetime values for the 'logs' and 'weeklySiteStats' datasets:

Datasets Definition file 'datasets.xml':

<datasets>  <dataset name="logs" frequency="${coord:hours(1)}"
           initial-instance="2009-01-01T01:00Z" timezone="UTC">
    <uri-template>hdfs://bar:8020/app/logs/${YEAR}${MONTH}/${DAY}/${HOUR}</uri-template>
  </dataset>
</datasets>

a. Coordinator application definition that creates a coordinator action once a day for a year, that is 365 coordinator actions:

   <coordinator-app name="app-coord" frequency="${coord:days(1)}"
                    start="2009-01-01T24:00Z" end="2009-12-31T24:00Z" timezone="UTC"
                    xmlns="uri:oozie:coordinator:0.1">
      <datasets>
        <include>hdfs://foo:8020/app/dataset-definitions/datasets.xml</include>
      </datasets>
      <input-events>
        <data-in name="input" dataset="logs">
          <start-instance>${coord:current(-23)}</start-instance>
          <end-instance>${coord:current(0)}</end-instance>
        </data-in>
      </input-events>
      <action>
        <workflow>
        ...
       </workflow>
      </action>      
   </coordinator-app>

Each coordinator action will require as input events the last 24 (-23 to 0) dataset instances for the 'logs' dataset. Because the dataset 'logs' is a hourly dataset, it means all its instances for the last 24 hours.

In this case, the dataset instances are used in a rolling window fashion.

b. Coordinator application definition that creates a coordinator action once an hour for a year, that is 8760 (24*8760) coordinator actions:

   <coordinator-app name="app-coord" frequency="${coord:hours(1)}"
                    start="2009-01-01T24:00Z" end="2009-12-31T24:00Z" timezone="UTC"
                    xmlns="uri:oozie:coordinator:0.1">
      <datasets>
        <include>hdfs://foo:8020/app/dataset-definitions/datasets.xml</include>
      </datasets>
      <input-events>
        <data-in name="input" dataset="logs">
          <start-instance>${coord:current(-23)}</start-instance>
          <end-instance>${coord:current(0)}</end-instance>
        </data-in>
      </input-events>
      <action>
        <workflow>
        ...
       </workflow>
      </action>      
   </coordinator-app>

Each coordinator action will require as input events the last 24 (-23 to 0) dataset instances for the 'logs' dataset. Similarly to the previous coordinator application example, it means all its instances for the last 24 hours.

However, because the frequency is hourly instead of daily, each coordinator action will use the last 23 dataset instances used by the previous coordinator action plus a new one.

In this case, the dataset instances are used in a sliding window fashion.

3. Using ${coord:current(int n)} to specify dataset instances created by a coordinator application:

Datasets Definition file 'datasets.xml':

<datasets>
.
  <dataset name="logs" frequency="${coord:hours(1)}"
           initial-instance="2009-01-01T01:00Z" timezone="UTC">
    <uri-template>hdfs://bar:8020/app/logs/${YEAR}/${MONTH}/${DAY}/${HOUR}</uri-template>
  </dataset>
.
  <dataset name="stats" frequency="${coord:days(1)}"
           initial-instance="2009-01-01T24:00Z" timezone="UTC">
    <uri-template>hdfs://bar:8020/app/logs/${YEAR}/${MONTH}/${DAY}</uri-template>
  </dataset>
.
</datasets>

Coordinator application definition:

   <coordinator-app name="app-coord" frequency="${coord:days(1)}"
                    start="2009-01-01T24:00Z" end="2009-12-31T24:00Z" timezone="UTC"
                    xmlns="uri:oozie:coordinator:0.1">
      <datasets>
        <include>hdfs://foo:8020/app/dataset-definitions/datasets.xml</include>
      </datasets>
      <input-events>
        <data-in name="input" dataset="logs">
          <start-instance>${coord:current(-23)}</start-instance>
          <end-instance>${coord:current(0)}</end-instance>
        </data-in>
      </input-events>
      <output-events>
        <data-out name="output" dataset="stats">
          <instance>${coord:current(0)}</instance>
        </data-out>
      </output-events>
      <action>
        <workflow>
        ...
       </workflow>
      </action>      
   </coordinator-app>

This coordinator application creates a coordinator action once a day for a year, this is 365 coordinator actions.

Each coordinator action will require as input events the last 24 (-23 to 0) dataset instances for the 'logs' dataset.

Each coordinator action will create as output event a new dataset instance for the 'stats' dataset.

Note that the 'stats' dataset initial-instance and frequency match the coordinator application start and frequency.

4. Using ${coord:current(int n)} to create a data-pipeline using a coordinator application:

This example shows how to chain together coordinator applications to create a data pipeline.

Dataset definitions file 'datasets.xml':

   <!--- Dataset A - produced every 15 minutes. -->
.
  <dataset name="15MinLogs" frequency="${coord:minutes(15)}"
           initial-instance="2009-01-01T00:15:00Z" timezone="UTC">
    <uri-template>hdfs://bar:8020/app/logs/${YEAR}/${MONTH}/${DAY}/${HOUR}/${MINUTE}</uri-template>
  </dataset>
.
  <dataset name="1HourLogs" frequency="${coord:hours(1)}"
           initial-instance="2009-01-01T01:00:00Z" timezone="UTC">
    <uri-template>hdfs://bar:8020/app/logs/${YEAR}/${MONTH}/${DAY}/${HOUR}</uri-template>
  </dataset>
.
  <dataset name="1DayLogs" frequency="${coord:hours(24)}"
           initial-instance="2009-01-01T24:00:00Z" timezone="UTC">
    <uri-template>hdfs://bar:8020/app/logs/${YEAR}/${MONTH}/${DAY}</uri-template>
  </dataset>

Coordinator application definitions. A data-pipeline with two coordinator-applications, one scheduled to run every hour, and another scheduled to run every day:

   <coordinator-app name="app-coord-hourly" frequency="${coord:hours(1)}"
                    start="2009-01-01T01:00Z" end="2009-12-31T24:00Z" timezone="UTC"
                    xmlns="uri:oozie:coordinator:0.1">
      <datasets>
        <include>hdfs://foo:8020/app/dataset-definitions/datasets.xml</include>
      </datasets>
      <input-events>
        <data-in name="input" dataset="15MinLogs">
          <start-instance>${coord:current(-3)}</start-instance>
          <end-instance>${coord:current(0)}</end-instance>
        </data-in>
      </input-events>
      <output-events>
        <data-out name="output" dataset="1HourLogs">
          <instance>${coord:current(0)}</instance>
        </data-out>
      </output-events>
      <action>
        <workflow>
        ...
       </workflow>
      </action>
   </coordinator-app>

   <coordinator-app name="app-coord-daily" frequency="${coord:days(1)}"
                    start="2009-01-01T24:00Z" end="2009-12-31T24:00Z" timezone="UTC"
                    xmlns="uri:oozie:coordinator:0.1">
      <datasets>
        <include>hdfs://foo:8020/app/dataset-definitions/datasets.xml</include>
      </datasets>
      <input-events>
        <data-in name="input" dataset="1HourLogs">
          <start-instance>${coord:current(-23)}</start-instance>
          <end-instance>${coord:current(0)}</end-instance>
        </data-in>
      </input-events>
      <output-events>
        <data-out name="output" dataset="1DayLogs">
          <instance>${coord:current(0)}</instance>
        </data-out>
      </output-events>
      <action>
        <workflow>
        ...
       </workflow>
      </action>      
   </coordinator-app>

The 'app-coord-hourly' coordinator application runs every every hour, uses 4 instances of the dataset "15MinLogs" to create one instance of the dataset "1HourLogs"

The 'app-coord-daily' coordinator application runs every every day, uses 24 instances of "1HourLogs" to create one instance of "1DayLogs"

The output datasets from the 'app-coord-hourly' coordinator application are the input to the 'app-coord-daily' coordinator application thereby forming a simple data-pipeline application.

6.6.2. coord:offset(int n, String timeUnit) EL Function for Synchronous Datasets

${coord:offset(int n, String timeUnit)} represents the n^th timeUnit, relative to the coordinator action creation (materialization) time. The coordinator action creation (materialization) time is computed based on the coordinator job start time and its frequency.

It is an alternative to the ${coord:current(int n)} command (see previous section) and can be used anywhere =${coord:current(int n)}= can be used. The difference between the two functions is that ${coord:current(int n)} computes an offset based on the n^th multiple of the frequency, while ${coord:offset(int n, String timeUnit)} computes an offset based on the n^th multiple of timeUnit .

n can be a negative integer, zero or a positive integer.

timeUnit can be any one of the following constants: "MINUTE" , "HOUR" , "DAY" , "MONTH" , "YEAR"

${coord:offset(int n, String timeUnit)} returns the nominal datetime for n^th timeUnit relative to the coordinator action creation (materialization) time.

When used directly, ${coord:offset(int n, String timeUnit)} performs the following calculation:

DS_FREQ: dataset frequency (minutes)
CA_NT = coordinator action creation (materialization) nominal time
coord:offset(int n, String timeUnit) = CA_NT + floor(timeUnit * n div DS_FREQ) * DS_FREQ

NOTE: The formula above is not 100% correct, because DST changes the calculation has to account for hour shifts. Oozie Coordinator must make the correct calculation accounting for DTS hour shifts.

When used in 'instance' or 'end-instance' XML elements, the above equation is used; the effect of the floor function is to "rewind" the resolved datetime to match the latest instance before the resolved time. When used in 'start-instance' XML elements, a slight modification to the above equation is used; instead of being "rewinded", the resolved datetime is "fastforwarded" to match the earliest instance after the resolved time. See the next two examples for more information.

Examples:

1. *=${coord:offset(int n, String timeUnit)}= datetime calculation:*

Datasets Definition:

<datasets>
.
  <dataset name="logs" frequency="${coord:days(1)}"
           initial-instance="2009-01-01T24:00Z" timezone="UTC">
    <uri-template>hdfs://bar:8020/app/logs/${YEAR}${MONTH}/${DAY}</uri-template>
  </dataset>
.
  <dataset name="weeklySiteAccessStats" frequency="${coord:days(7)}"
           initial-instance="2009-01-07T24:00Z" timezone="UTC">
    <uri-template>hdfs://bar:8020/app/weeklystats/${YEAR}/${MONTH}/${DAY}</uri-template>
  </dataset>
.
</datasets>

For a coordinator action creation time: 2009-05-29T24:00Z the ${coord:offset(int n, String timeUnit)} EL function would resolve to the following datetime values for the 'logs' and 'weeklySiteStats' datasets:

1. When n is 0, the timeUnit doesn't really matter because 0 minutes is the same as 0 hours, 0 days, etc
2. There are multiple ways to express the same value (e.g. ${coord:offset(24, "HOUR")}= is equivalent to ${coord:offset(1, "DAY")} )
3. The datetimes resolved for the 2 datasets differ when the ${coord:offset(int n, String timeUnit)} function is invoked with the same arguments. This is because the ${coord:offset(int n, String timeUnit)} function takes into consideration the initial-time and the frequency for the dataset for which is performing the calculation.
4. As mentioned before, if the resolved time doesn't fall exactly on an instance, it will get "rewinded" to match the latest instance before the resolved time. For example, ${coord:offset(1, "DAY")} is resolved to 2009-05-27T24:00Z for the 'weeklysiteStats' dataset even though this is the same as ${coord:offset(0, "DAY")} ; this is because the frequency is 7 days, so ${coord:offset(1, "DAY")} had to be "rewinded".

2. "fastforwarding" in ${coord:offset(int n, String timeUnit)} calculation:

When specifying dataset instances, keep in mind that the resolved value of ${coord:offset(int n, String timeUnit)} must line up with an offset of a multiple of the frequency when used in an 'instance' XML element. However, when used in 'start-instance' and 'end-instance' XML elements, this is not a requirement. In this case, the function will automatically resolve the range of instances to match the offset of a multiple of the frequency that would fall between the ='start-instance'= and 'end-instance' XML elements; in other words, 'start-instance' XML element is "fastforwarded" while ='end-instance'= XML element is "rewinded". So, in the example below, the frequency of the "logs" dataset is 1 hour while the ='start-instance'= XML element is ${coord:offset(-90, "MINUTE")} (-1.5 hours). If this were in an 'instance' XML element, it would be "rewinded", but here it is effectively equivalent to ${coord:offset(-60, "MINUTE")} or ${coord:current(-1)} as we are dealing with a range.

Datasets Definition file 'datasets.xml':

<datasets>
.
  <dataset name="logs" frequency="${coord:hours(1)}"
           initial-instance="2009-01-01T01:00Z" timezone="UTC">
    <uri-template>hdfs://bar:8020/app/logs/${YEAR}/${MONTH}/${DAY}/${HOUR}</uri-template>
  </dataset>
.
  <dataset name="stats" frequency="${coord:days(1)}"
           initial-instance="2009-01-01T24:00Z" timezone="UTC">
    <uri-template>hdfs://bar:8020/app/logs/${YEAR}/${MONTH}/${DAY}</uri-template>
  </dataset>
.
</datasets>

Coordinator application definition:

   <coordinator-app name="app-coord" frequency="${coord:days(1)}"
                    start="2009-01-01T24:00Z" end="2009-12-31T24:00Z" timezone="UTC"
                    xmlns="uri:oozie:coordinator:0.1">
      <datasets>
        <include>hdfs://foo:8020/app/dataset-definitions/datasets.xml</include>
      </datasets>
      <input-events>
        <data-in name="input" dataset="logs">
          <start-instance>${coord:offset(-90, "MINUTE")}</start-instance>
          <end-instance>${coord:offset(0, "DAY")}</end-instance>
        </data-in>
      </input-events>
      <output-events>
        <data-out name="output" dataset="stats">
          <instance>${coord:offset(0, "DAY")}</instance>
        </data-out>
      </output-events>
      <action>
        <workflow>
        ...
       </workflow>
      </action>
   </coordinator-app>

6.6.3. coord:hoursInDay(int n) EL Function for Synchronous Datasets

The ${coord:hoursInDay(int n)} EL function returns the number of hours for the specified day, in a timezone/daylight-saving sensitive way.

n is offset (in days) from the current nominal time. A negative value is the n^th previous day. Zero is the current day. A positive number is the n^th next day.

The returned value is calculated taking into account timezone daylight-saving information.

Normally it returns 24 , only DST switch days for the timezone in question it will return either 23 or 25 .

Examples:

Coordinator application definition:

   <coordinator-app name="app-coord" frequency="${coord:days(1)}"
                    start="${jobStart}" end="${jobEnd}" timezone="${timezone}"
                    xmlns="uri:oozie:coordinator:0.1">
      <datasets>
        <dataset name="logs" frequency="${coord:hours(1)}"
                 initial-instance="${logsInitialInstance}" timezone="${timezone}">
          <uri-template>
            hdfs://bar:8020/app/logs/${market}/${language}/${YEAR}${MONTH}/${DAY}/${HOUR}
          </uri-template>
        </dataset>
      </datasets>
      <input-events>
        <data-in name="input" dataset="logs">
          <start-instance>${coord:current( -(coord:hoursInDay(0) - 1) )}</start-instance>
          <end-instance>${coord:current(0)}</end-instance>
        </data-in>
      </input-events>
      <action>
        <workflow>
        ...
       </workflow>
      </action>
   </coordinator-app>

This example is the example of section #6.5 but with a minor change. The argument for the ${coord:current(int n)} function in the 'start-instance' element, instead using -23 , the example now uses -(coord:hoursInDay(0) - 1) .

This simple change fully enables this coordinator application to handle daily data (produced hourly) for any timezone, with timezones observing or not daylight saving.

For timezones observing daylight saving, on the days of DST switch, the function will resolve to 23 or 25 , thus the dataset instances used will be for for the day in the DST sense.

For timezones not observing daylight saving, it always returns 24 .

6.6.4. coord:daysInMonth(int n) EL Function for Synchronous Datasets

The ${coord:daysInMonth(int n)} EL function returns the number of days for month of the specified day.

n is offset (in months) from the current nominal time. A negative value is the n^th previous month. Zero is the current month. A positive number is the n^th next month.

The returned value is calculated taking into account leap years information.

The ${coord:daysInMonth(int n)} EL function can be used to express monthly ranges for dataset instances.

Examples:

Coordinator application definition:

   <coordinator-app name="app-coord" frequency="${coord:months(1)}"
                    start="2009-01-31T24:00Z" end="2009-12-31T24:00" timezone="UTC"
                    xmlns="uri:oozie:coordinator:0.1">
      <datasets>
        <dataset name="logs" frequency="${coord:days(1)}"
                 initial-instance="2009-01-01T24:00Z" timezone="UTC">
          <uri-template>
            hdfs://bar:8020/app/logs/${market}/${language}/${YEAR}${MONTH}/${DAY}
          </uri-template>
        </dataset>
      </datasets>
      <input-events>
        <data-in name="input" dataset="logs">
          <start-instance>${coord:current( -(coord:daysInMonth(0) - 1) )}</start-instance>
          <end-instance>${coord:current(0)}</end-instance>
        </data-in>
      </input-events>
      <action>
        <workflow>
        ...
       </workflow>
      </action>
   </coordinator-app>

This example is a coordinator application that runs monthly, and consumes the daily feeds for the last month.

6.6.5. coord:tzOffset() EL Function for Synchronous Datasets

${coord:tzOffset()} EL function returns the difference in minutes between a dataset timezone and the coordinator job timezone at the current nominal time. This EL function is useful when dealing with datasets from multiple timezones, but execute in a different timezone.

  DS_TZ : dataset TZ offset in minutes at the current nominal time (UTC offset)
  JOB_TZ: coordinator job UTC TZ offset in minutes at the current nominal time (UTC offset).  coord:tzOffset() = DS_TZ - JOB_TZ

For example: Los Angeles Winter offset (no DST) is -480 (-08:00 hours). India offset is -330 (+05:30 hours).

The value returned by this function may change because of the daylight saving rules of the 2 timezones. For example, between Continental Europe and The U.S. West coast, most of the year the timezone different is 9 hours, but there are a few day or weeks.

IMPORTANT: While the offset is multiples of 60 for most timezones, it can be multiple of 30 mins when one of the timezones is has a ##:30 offset (i.e. India).

Refer to section #7, 3nd use case for a detailed example.

6.6.6. coord:latest(int n) EL Function for Synchronous Datasets

${coord:latest(int n)} represents the n^th latest currently available instance of a synchronous dataset.

${coord:latest(int n)} is not relative to the coordinator action creation (materialization) time, it is the n^th latest instance available when the action is started (when the workflow job is started).

If a coordinator job is suspended, when resumed, all usages of ${coord:latest(int n)} will be resolved to the currently existent instances.

Finally, it is not possible to represent the latest dataset when execution reaches a node in the workflow job. The resolution of latest dataset instances happens at action start time (workflow job start time).

The parameter n can be a negative integer or zero. Where 0 means the latest instance available, -1 means the second latest instance available, etc.

the ${coord:latest(int n)} ignores gaps in dataset instances, it just looks for the latest n^th instance available.

Example: :

Coordinator application definition:

   <coordinator-app name="app-coord" frequency="${coord:hours(1)}"
                    start="2009-01-01T24:00Z" end="2009-12-31T24:00Z" timezone="UTC"
                    xmlns="uri:oozie:coordinator:0.1">
      <datasets>
        <dataset name="logs" frequency="${coord:hours(1)}"
                 initial-instance="2009-01-01T01:00Z" timezone="UTC">
          <uri-template>
            hdfs://bar:8020/app/logs/${YEAR}/${MONTH}/${DAY}/${HOUR}
          </uri-template>
        </dataset>
      </datasets>
      <input-events>
        <data-in name="input" dataset="logs">
          <instance>${coord:latest(-2)}</instance>
          <instance>${coord:latest(0)}</instance>
        </data-in>
      </input-events>
      <action>
        <workflow>
        ...
       </workflow>
      </action>      
   </coordinator-app>

If the available dataset instances in HDFS at time of a coordinator action being executed are:

  hdfs://bar:8020/app/logs/2009/01/01
  hdfs://bar:8020/app/logs/2009/01/02
  hdfs://bar:8020/app/logs/2009/01/03
  	(missing)
  hdfs://bar:8020/app/logs/2009/01/05
  (missing)
  hdfs://bar:8020/app/logs/2009/01/07
  (missing)
  (missing)
  hdfs://bar:8020/app/logs/2009/01/10

Then, the dataset instances for the input events for the coordinator action will be:

  hdfs://bar:8020/app/logs/2009/01/05
  hdfs://bar:8020/app/logs/2009/01/10

6.6.7. coord:future(int n, int limit) EL Function for Synchronous Datasets

${coord:future(int n, int limit)} represents the n^th currently available future instance of a synchronous dataset while looking ahead for 'limit' number of instances.

${coord:future(int n, int limit)} is relative to the coordinator action creation (materialization) time. The coordinator action creation (materialization) time is computed based on the coordinator job start time and its frequency. The n^th dataset instance is computed based on the dataset's initial-instance datetime, its frequency and the (current) coordinator action creation (materialization) time.

n can be a zero or a positive integer. Where 0 means the immediate instance available, 1 means the second next instance available, etc.

limit should be a positive integer. Where 3 means search for n^th next instance and should not check beyond 3 instance.

The ${coord:future(int n, int limit)} ignores gaps in dataset instances, it just looks for the next n^th instance available.

Example: :

Coordinator application definition:

   <coordinator-app name="app-coord" frequency="${coord:hours(1)}"
                    start="2009-01-01T24:00Z" end="2009-12-31T24:00Z" timezone="UTC"
                    xmlns="uri:oozie:coordinator:0.1">
      <datasets>
        <dataset name="logs" frequency="${coord:hours(1)}"
                 initial-instance="2009-01-01T01:00Z" timezone="UTC">
          <uri-template>
            hdfs://bar:8020/app/logs/${YEAR}/${MONTH}/${DAY}/${HOUR}
          </uri-template>
        </dataset>
      </datasets>
      <input-events>
        <data-in name="input" dataset="logs">
          <instance>${coord:future(0, 10)}</instance>
          <instance>${coord:future(2, 10)}</instance>
        </data-in>
      </input-events>
      <action>
        <workflow>
        ...
       </workflow>
      </action>
   </coordinator-app>

If the available dataset instances in HDFS at time of a coordinator action being executed are:

  hdfs://bar:8020/app/logs/2009/02/01
  (missing)
  (missing)
  (missing)
  hdfs://bar:8020/app/logs/2009/02/04
 (missing)
 (missing)
  hdfs://bar:8020/app/logs/2009/02/07
  (missing)
  (missing)
  (missing)
  hdfs://bar:8020/app/logs/2009/02/11
  (missing)
  (missing)
  hdfs://bar:8020/app/logs/2009/02/14
  (missing)
  hdfs://bar:8020/app/logs/2009/02/16

Then, the dataset instances for the input events for the coordinator action will be:

  hdfs://bar:8020/app/logs/2009/02/01
  hdfs://bar:8020/app/logs/2009/02/07

6.6.8. coord:version(int n) EL Function for Asynchronous Datasets

• TBD

6.6.9. coord:latest(int n) EL Function for Asynchronous Datasets

• TBD

6.6.10. Dataset Instance Resolution for Instances Before the Initial Instance

When defining input events that refer to dataset instances it may be possible that the resolution of instances is out of it lower bound. This is scenario is likely to happen when the instance resolution is very close to the initial-instance. This is useful for bootstrapping the application.

To address this edge scenario, Oozie Coordinator silently ignores dataset instances out of bounds.

Example: :

Coordinator application definition:

   <coordinator-app name="app-coord" frequency="${coord:hours(1)}"
                    start="2009-01-01T01:00Z" end="2009-12-31T24:00Z" timezone="UTC"
                    xmlns="uri:oozie:coordinator:0.1">
      <datasets>
        <dataset name="logs" frequency="${coord:hours(1)}"
                 initial-instance="2009-01-01T00:00Z"  timezone="UTC">
          <uri-template>
            hdfs://bar:8020/app/logs/${YEAR}/${MONTH}/${DAY}/${HOUR}
          </uri-template>
        </dataset>
      </datasets>
      <input-events>
        <data-in name="inputLogs" dataset="logs">
          <start-instance>${coord:current(-23)}</start-instance>
          <end-instance>${coord:current(0)}</end-instance>
        </data-in>
      </input-events>
      <action>
        <workflow>
        ...
       </workflow>
      </action>
   </coordinator-app>

In the case of the synchronous 'logs' dataset, for the first action of this coordinator job, the instances referred in the input events will resolve to just 1 instance. For the second action it will resolve to 2 instances. And so on. Only after the 24th action, the input events will resolve constantly to 24 instances. In other words, while ${coord:current(-23)} resolves to datetimes prior to the 'initial-instance' the required range will start from the 'initial-instance', '2009-01-01T00:00Z' in this example.

6.7.1. coord:dataIn(String name) EL Function

The ${coord:dataIn(String name)} EL function resolves to all the URIs for the dataset instances specified in an input event dataset section.

The ${coord:dataIn(String name)} is commonly used to pass the URIs of dataset instances that will be consumed by a workflow job triggered by a coordinator action.

Example: :

Coordinator application definition:

   <coordinator-app name="app-coord" frequency="${coord:days(1)}"
                    start="2009-01-01T24:00Z" end="2009-12-31T24:00Z" timezone="UTC"
                    xmlns="uri:oozie:coordinator:0.1">
      <datasets>
        <dataset name="logs" frequency="${coord:hours(1)}"
                 initial-instance="2009-01-01T01:00Z" timezone="UTC">
          <uri-template>
             hdfs://bar:8020/app/logs/${YEAR}/${MONTH}/${DAY}/${HOUR}
          </uri-template>
        </dataset>
      </datasets>
      <input-events>
        <data-in name="inputLogs" dataset="logs">
          <start-instance>${coord:current( -(coord:hoursInDay(0) - 1) )}</start-instance>
          <end-instance>${coord:current(-1)}</end-instance>
        </data-in>
      </input-events>
      <action>
        <workflow>
          <app-path>hdfs://bar:8020/usr/joe/logsprocessor-wf</app-path>
          <configuration>
            <property>
              <name>wfInput</name>
              <value>${coord:dataIn('inputLogs')}</value>
            </property>
         </configuration>
       </workflow>
      </action>
   </coordinator-app>

In this example, each coordinator action will use as input events the the last day hourly instances of the 'logs' dataset.

The ${coord:dataIn(String name)} function enables the coordinator application to pass the URIs of all the dataset instances for the last day to the workflow job triggered by the coordinator action. For the 2009-01-02T00:00Z" run, the =${coord:dataIn('inputLogs')} function will resolve to:

  hdfs://bar:8020/app/logs/2009/01/01/01,
  hdfs://bar:8020/app/logs/2009/01/01/02,
  ...
  hdfs://bar:8020/app/logs/2009/01/01/23,
  hdfs://bar:8020/app/logs/2009/02/00/00

The ${coord:dataIn('inputLogs')} is used for workflow job configuration property 'wfInput' for the workflow job that will be submitted by the coordinator action on January 2nd 2009. Thus, when the workflow job gets started, the 'wfInput' workflow job configuration property will contain all the above URIs.

Note that all the URIs form a single string value and the URIs are separated by commas. Multiple HDFS URIs separated by commas can be specified as input data to a Map/Reduce job.

6.7.2. coord:dataOut(String name) EL Function

The ${coord:dataOut(String name)} EL function resolves to all the URIs for the dataset instance specified in an output event dataset section.

The ${coord:dataOut(String name)} is commonly used to pass the URIs of a dataset instance that will be produced by a workflow job triggered by a coordinator action.

Example: :

Datasets Definition file 'datasets.xml'

<datasets>
.
  <dataset name="hourlyLogs" frequency="${coord:hours(1)}"
           initial-instance="2009-01-01T01:00Z" timezone="UTC">
    <uri-template>hdfs://bar:8020/app/logs/${YEAR}/${MONTH}/${DAY}/${HOUR}</uri-template>
  </dataset>
.
  <dataset name="dailyLogs" frequency="${coord:days(1)}"
           initial-instance="2009-01-01T24:00Z" timezone="UTC">
    <uri-template>hdfs://bar:8020/app/daily-logs/${YEAR}/${MONTH}/${DAY}</uri-template>
  </dataset>
</datasets>

Coordinator application definition:

   <coordinator-app name="app-coord" frequency="${coord:days(1)}"
                    start="2009-01-01T24:00Z" end="2009-12-31T24:00Z" timezone="UTC"
                    xmlns="uri:oozie:coordinator:0.1">
      <datasets>
        <include>hdfs://foo:8020/app/dataset-definitions/datasets.xml</include>
      </datasets>
      <input-events>
        <data-in name="inputLogs" dataset="hourlyLogs">
          <start-instance>${coord:current( -(coord:hoursInDay(0) -1) )}</start-instance>
          <end-instance>${coord:current(0)}</end-instance>
        </data-in>
      </input-events>
      <output-events>
        <data-out name="outputLogs" dataset="dailyLogs">
          <instance>${coord:current(0)}</instance>
        </data-out>
      </output-events>
      <action>
        <workflow>
          <app-path>hdfs://bar:8020/usr/joe/logsaggretor-wf</app-path>
          <configuration>
            <property>
              <name>wfInput</name>
              <value>${coord:dataIn('inputLogs')}</value>
            </property>
            <property>
              <name>wfOutput</name>
              <value>${coord:dataOut('outputLogs')}</value>
            </property>
         </configuration>
       </workflow>
      </action>
   </coordinator-app>

In this example, each coordinator action will use as input events the the last 24 hourly instances of the 'hourlyLogs' dataset to create a 'dailyLogs' dataset instance.

The ${coord:dataOut(String name)} function enables the coordinator application to pass the URIs of the the dataset instance that will be created by the workflow job triggered by the coordinator action. For the 2009-01-01T24:00Z" run, the =${coord:dataOut('dailyLogs')} function will resolve to:

  hdfs://bar:8020/app/logs/2009/01/02

NOTE: The use of 24:00 as hour is useful for human to denote end of the day, but internally Oozie handles it as the zero hour of the next day.

The ${coord:dataOut('dailyLogs')} is used for workflow job configuration property 'wfOutput' for the workflow job that will be submitted by the coordinator action on January 2nd 2009. Thus, when the workflow job gets started, the 'wfOutput' workflow job configuration property will contain the above URI.

6.7.3. coord:nominalTime() EL Function

The ${coord:nominalTime()} EL function resolves to the coordinator action creation (materialization) datetime.

The nominal times is always the coordinator job start datetime plus a multiple of the coordinator job frequency.

This is, when the coordinator action was created based on driver event. For synchronous coordinator applications this would be every tick of the frequency.

Example: :

Coordinator application definition:

   <coordinator-app name="app-coord" frequency="${coord:days(1)}"
                    start="2009-01-01T24:00Z" end="2009-12-31T24:00Z" timezone="UTC"
                    xmlns="uri:oozie:coordinator:0.1">
     <datasets>
       <dataset name="hourlyLogs" frequency="${coord:hours(1)}"
                initial-instance="2009-01-01T01:00Z" timezone="UTC">
         <uri-template>hdfs://bar:8020/app/logs/${YEAR}/${MONTH}/${DAY}/${HOUR}</uri-template>
       </dataset>
     </datasets>
      <input-events>
        <data-in name="inputLogs" dataset="hourlyLogs">
          <start-instance>${coord:current(-23)}</start-instance>
          <end-instance>${coord:current(0)}</end-instance>
        </data-in>
      </input-events>
      <action>
        <app-path>${nameNode}/user/${coord:user()}/examples/apps/aggregator</app-path>
        <configuration>
            <property>
              <name>nextInstance</name>
              <value>${coord:dateOffset(coord:nominalTime(), 1, 'DAY')}</value>
            </property>
            <property>
             <name>previousInstance</name>
              <value>${coord:dateOffset(coord:nominalTime(), -1, 'DAY')}</value>
            </property>
         </configuration>
      </action>
   </coordinator-app>

The nominal times for the coordinator actions of this coordinator application example are:

  2009-01-02T00:00Z
  2009-01-03T00:00Z
  2009-01-04T00:00Z
  ...
  2010-01-01T00:00Z

These are the times the action where created (materialized).

6.7.4. coord:actualTime() EL Function

The ${coord:actualTime()} EL function resolves to the coordinator action actual creation datetime.

When the coordinator action is created based on driver event, the current time is saved to action. An action's actual time is less than the nominal time if coordinator job is in running in current mode. If job is running as catch-up mode (job's start time is in the past), the actual time is greater than the nominal time.

Example: :

Coordinator application definition:

   <coordinator-app name="app-coord" frequency="${coord:days(1)}"
                    start="2011-05-01T24:00Z" end="2011-12-31T24:00Z" timezone="UTC"
                    xmlns="uri:oozie:coordinator:0.1">
     <datasets>
       <dataset name="hourlyLogs" frequency="${coord:hours(1)}"
                initial-instance="2011-04-01T01:00Z" timezone="UTC">
         <uri-template>hdfs://bar:8020/app/logs/${YEAR}/${MONTH}/${DAY}/${HOUR}</uri-template>
       </dataset>
     </datasets>
      <input-events>
        <data-in name="inputLogs" dataset="hourlyLogs">
          <start-instance>${coord:current(-23)}</start-instance>
          <end-instance>${coord:current(0)}</end-instance>
        </data-in>
      </input-events>
      <action>
        <app-path>${nameNode}/user/${coord:user()}/examples/apps/aggregator</app-path>
        <configuration>
            <property>
              <name>actualTime</name>
              <value>${coord:formatTime(coord:actualTime(), 'yyyy-MM-dd')}</value>
            </property>
         </configuration>
      </action>
   </coordinator-app>

The actual time for the coordinator actions of this coordinator application example will look like:

If coordinator job was started at 2011-05-01, then actions' actualTime is

  2011-05-01
  2011-05-02
  2011-05-03
  ...
  2011-12-31

6.7.5. coord:user() EL Function (since Oozie 2.3)

The coord:user() function returns the user that started the coordinator job.

6.8.1 coord:databaseIn(String name), coord:databaseOut(String name) EL function

The functions ${coord:databaseIn(String name} and ${coord:databaseOut(String name} are used to pass the database name of HCat dataset instances, input and output respectively, that will be consumed by a workflow job triggered by a coordinator action.

These functions take as argument - name of the dataset and give as string the 'database' name of that dataset. If dataset belongs to 'input-events', use ${coord:databaseIn(String name} , else if it belongs to 'output-events', use ${coord:databaseOut(String name} .

Refer to the Example below for usage.

6.8.2 coord:tableIn(String name), coord:tableOut(String name) EL function

The functions ${coord:tableIn(String name} and ${coord:tableOut(String name} are used to pass the table name of HCat dataset instances, input and output respectively, that will be consumed by a workflow job triggered by a coordinator action.

These functions take as argument - name of the dataset and give as string the 'table' name of that dataset. If dataset belongs to 'input-events', use ${coord:tableIn(String name} , else if it belongs to 'output-events', use ${coord:tableOut(String name} .

Refer to the Example below for usage.

6.8.3 coord:dataInPartitionFilter(String name, String type) EL function

The ${coord:dataInPartitionFilter(String name, String type)} EL function resolves to a filter clause to filter all the partitions corresponding to the dataset instances specified in an input event dataset section. This EL function takes two arguments - the name of the input dataset, and the type of the workflow action which will be consuming this filter. There are 3 types - 'pig', 'hive' and 'java'. This filter clause from the EL function is to be passed as a parameter in the respective action in the workflow.

The evaluated value of the filter clause will vary based on the action type passed to the EL function. In case of pig, the filter will have "==" as the equality operator in the condition. In case of hive and java, the filter will have "=" as the equality operator in the condition. The type java is for java actions, which use HCatInputFormat directly and launch jobs. The filter clause in that case can be used to construct the InputJobInfo in =HCatInputFormat.setInput(Job job, InputJobInfo inputJobInfo)=.

Refer to the Example below for usage.

6.8.4 coord:dataOutPartitions(String name) EL function

The ${coord:dataOutPartitions(String name)} EL function resolves to a comma-separated list of partition key-value pairs for the output-event dataset. This can be passed as an argument to HCatStorer in Pig scripts or in case of java actions that directly use HCatOutputFormat and launch jobs, the partitions list can be parsed to construct partition values map for OutputJobInfo in HcatOutputFormat.setOutput(Job job, OutputJobInfo outputJobInfo) .

The example below illustrates a pig job triggered by a coordinator, using the EL functions for HCat database, table, input partitions filter and output partitions. The example takes as input previous day's hourly data to produce aggregated daily output.

Example:

Coordinator application definition:

   <coordinator-app name="app-coord" frequency="${coord:days(1)}"
                    start="2009-01-01T24:00Z" end="2009-12-31T24:00Z" timezone="UTC"
                    xmlns="uri:oozie:coordinator:0.3">
      <datasets>
        <dataset name="Click-data" frequency="${coord:hours(1)}"
                 initial-instance="2009-01-01T01:00Z" timezone="UTC">
          <uri-template>
             hcat://foo:11002/myInputDatabase/myInputTable/datestamp=${YEAR}${MONTH}${DAY}${HOUR};region=USA
          </uri-template>
        </dataset>
        <dataset name="Stats" frequency="${coord:days(1)}"
                 initial-instance="2009-01-01T01:00Z" timezone="UTC">
          <uri-template>
             hcat://foo:11002/myOutputDatabase/myOutputTable/datestamp=${YEAR}${MONTH}${DAY}
          </uri-template>
        </dataset>
      </datasets>
      <input-events>
        <data-in name="raw-logs" dataset="Click-data">
          <start-instance>${coord:current(-23)}</start-instance>
          <end-instance>${coord:current(0)}</end-instance>
        </data-in>
      </input-events>
      <output-events>
        <data-out name="processed-logs" dataset="Stats">
          <instance>${coord:current(0)}</instance>
        </data-out>
      </output-events>
      <action>
        <workflow>
          <app-path>hdfs://bar:8020/usr/joe/logsprocessor-wf</app-path>
          <configuration>
            <property>
              <name>IN_DB</name>
              <value>${coord:databaseIn('Click-data')}</value>
            </property>
            <property>
              <name>IN_TABLE</name>
              <value>${coord:tableIn('raw-logs')}</value>
            </property>
            <property>
              <name>FILTER</name>
              <value>${coord:dataInPartitionFilter('raw-logs', 'pig')}</value>
            </property>
            <property>
              <name>OUT_DB</name>
              <value>${coord:databaseOut('processed-logs')}</value>
            </property>
            <property>
              <name>OUT_TABLE</name>
              <value>${coord:tableOut('processed-logs')}</value>
            </property>
            <property>
              <name>OUT_PARTITIONS</name>
              <value>${coord:dataOutPartitions('processed-logs')}</value>
            </property>
         </configuration>
       </workflow>
      </action>
   </coordinator-app>

Parameterizing the input/output databases and tables using the corresponding EL function as shown will make them available in the pig action of the workflow 'logsprocessor-wf'.

Each coordinator action will use as input events the last 24 hourly instances of the 'Click-data' dataset. The ${coord:dataInPartitionFilter(String name, String type)} function enables the coordinator application to pass the Partition Filter corresponding to all the dataset instances for the last 24 hours to the workflow job triggered by the coordinator action. The ${coord:dataOutPartitions(String name)} function enables the coordinator application to pass the partition key-value string needed by the HCatStorer in Pig job when the workflow is triggered by the coordinator action.

Workflow definition:

<workflow-app xmlns="uri:oozie:workflow:0.3" name="logsprocessor-wf">
    <credentials>
      <credential name='hcatauth' type='hcat'>
        <property>
          <name>hcat.metastore.uri</name>
          <value>${HCAT_URI}</value>
        <property>
        </property>
          <name>hcat.metastore.principal</name>
          <value>${HCAT_PRINCIPAL}</value>
        <property>
      </credential>
    </credentials>
    <start to="pig-node"/>
    <action name="pig-node" cred="hcatauth">
        <pig>
            <job-tracker>${jobTracker}</job-tracker>
            <name-node>${nameNode}</name-node>
            <prepare>
                <delete path="hcat://foo:11002/${OUT_DB}/${OUT_TABLE}/date=${OUT_PARTITION_VAL_DATE}"/>
            </prepare>
            ...
            <script>id.pig</script>
            <param>HCAT_IN_DB=${IN_DB}</param>
            <param>HCAT_IN_TABLE=${IN_TABLE}</param>
            <param>HCAT_OUT_DB=${OUT_DB}</param>
            <param>HCAT_OUT_TABLE=${OUT_TABLE}</param>
            <param>PARTITION_FILTER=${FILTER}</param>
            <param>OUTPUT_PARTITIONS=${OUT_PARTITIONS}</param>
        <file>lib/hive-site.xml</file>
        </pig>
        <ok to="end"/>
        <error to="fail"/>
    </action>
    <kill name="fail">
        <message>Pig failed, error message[${wf:errorMessage(wf:lastErrorNode())}]</message>
    </kill>
    <end name="end"/>
</workflow-app>

Ensure that the following jars are in classpath, with versions corresponding to hcatalog installation: hcatalog-core.jar, webhcat-java-client.jar, hive-common.jar, hive-exec.jar, hive-metastore.jar, hive-serde.jar, libfb303.jar and pig.jar. You can also specify the jars using archive tag. The hive-site.xml needs to be provided using file tag.

Example usage in Pig:

A = load '$HCAT_IN_DB.$HCAT_IN_TABLE' using org.apache.hcatalog.pig.HCatLoader();
B = FILTER A BY $PARTITION_FILTER;
C = foreach B generate foo, bar;
store C into '$HCAT_OUT_DB.$HCAT_OUT_TABLE' using org.apache.hcatalog.pig.HCatStorer('$OUTPUT_PARTITIONS');

For the 2009-01-02T00:00Z run with the given dataset instances, the above Pig script with resolved values would look like:

A  load 'myInputDatabase.myInputTable' using org.apache.hcatalog.pig.HCatLoader();
B = FILTER A BY ((datestamp
=2009010101 AND region==USA) OR
    (datestamp==2009010102 AND region==USA) OR
    ...
    (datestamp==2009010123 AND region==USA) OR
    (datestamp==2009010200 AND region==USA));
C = foreach B generate foo, bar;
store C into 'myOutputDatabase.myOutputTable' using org.apache.hcatalog.pig.HCatStorer('datestamp=20090102,region=EUR');

6.8.5 coord:dataInPartitionMin(String name, String partition) EL function

The ${coord:dataInPartitionMin(String name, String partition)} EL function resolves to the minimum value of the specified partition for all the dataset instances specified in an input event dataset section. It can be used to do range based filtering of partitions in pig scripts together with dataInPartitionMax EL function.

Refer to the Example below for usage.

6.8.6 coord:dataInPartitionMax(String name, String partition) EL function

The ${coord:dataInPartitionMax(String name, String partition)} EL function resolves to the maximum value of the specified partition for all the dataset instances specified in an input event dataset section. It is a better practice to use dataInPartitionMin and dataInPartitionMax to form a range filter wherever possible instead of datainPartitionPigFilter as it will be more efficient for filtering.

Refer to the Example below for usage.

6.8.7 coord:dataOutPartitionValue(String name, String partition) EL function

The ${coord:dataOutPartitionValue(String name, String partition)} EL function resolves to value of the specified partition for the output-event dataset; that will be consumed by a workflow job, e.g Pig job triggered by a coordinator action. This is another convenience function to use a single partition-key's value if required, in addition to dataoutPartitionsPig() and either one can be used.

The example below illustrates a pig job triggered by a coordinator, using the aforementioned EL functions for input partition max/min values, output partition value, and database and table.

Example:

Coordinator application definition:

   <coordinator-app name="app-coord" frequency="${coord:days(1)}"
                    start="2009-01-01T24:00Z" end="2009-12-31T24:00Z" timezone="UTC"
                    xmlns="uri:oozie:coordinator:0.1">
      <datasets>
        <dataset name="Click-data" frequency="${coord:hours(1)}"
                 initial-instance="2009-01-01T01:00Z" timezone="UTC">
          <uri-template>
             hcat://foo:11002/myInputDatabase/myInputTable/datestamp=${YEAR}${MONTH}${DAY}${HOUR};region=USA
          </uri-template>
        </dataset>
        <dataset name="Stats" frequency="${coord:days(1)}"
                 initial-instance="2009-01-01T01:00Z" timezone="UTC">
          <uri-template>
             hcat://foo:11002/myOutputDatabase/myOutputTable/datestamp=${YEAR}${MONTH}${DAY};region=USA
          </uri-template>
        </dataset>
      </datasets>
      <input-events>
        <data-in name="raw-logs" dataset="Click-data">
          <start-instance>${coord:current(-23)}</start-instance>
          <end-instance>${coord:current(0)}</end-instance>
        </data-in>
      </input-events>
      <output-events>
        <data-out name="processed-logs" dataset="Stats">
          <instance>${coord:current(0)}</instance>
        </data-out>
      </output-events>
      <action>
        <workflow>
          <app-path>hdfs://bar:8020/usr/joe/logsprocessor-wf</app-path>
          <configuration>
            <property>
              <name>IN_DB</name>
              <value>${coord:databaseIn('raw-logs')}</value>
            </property>
            <property>
              <name>IN_TABLE</name>
              <value>${coord:tableIn('raw-logs')}</value>
            </property>
            <property>
              <name>DATE_MIN</name>
              <value>${coord:dataInPartitionMin('raw-logs','datestamp')}</value>
            </property>
            <property>
              <name>DATE_MAX</name>
              <value>${coord:dataInPartitionMax('raw-logs','datestamp')}</value>
            </property>
            <property>
              <name>OUT_DB</name>
              <value>${coord:databaseOut('processed-logs')}</value>
            </property>
            <property>
              <name>OUT_TABLE</name>
              <value>${coord:tableOut('processed-logs')}</value>
            </property>
            <property>
              <name>OUT_PARTITION_VAL_REGION</name>
              <value>${coord:dataOutPartitionValue('processed-logs','region')}</value>
            </property>
            <property>
              <name>OUT_PARTITION_VAL_DATE</name>
              <value>${coord:dataOutPartitionValue('processed-logs','datestamp')}</value>
            </property>
         </configuration>
       </workflow>
      </action>
   </coordinator-app>

In this example, each coordinator action will use as input events the last 24 hourly instances of the 'logs' dataset.

For the 2009-01-02T00:00Z run, the ${coord:dataInPartitionMin('raw-logs','datestamp')} function will resolve to the minimum of the 5 dataset instances for partition 'datestamp' i.e. among 2009010101, 2009010102, ...., 2009010123, 2009010200, the minimum would be "2009010101".

Similarly, the ${coord:dataInPartitionMax('raw-logs','datestamp')} function will resolve to the maximum of the 5 dataset instances for partition 'datestamp' i.e. among 2009010120, 2009010121, ...., 2009010123, 2009010200, the maximum would be "2009010200".

Finally, the ${coord:dataOutPartitionValue(String name, String partition)} function enables the coordinator application to pass a specified partition's value string needed by the HCatStorer in Pig job. The ${coord:dataOutPartitionValue('processed-logs','region')} function will resolve to: "${region}" and ${coord:dataOutPartitionValue('processed-logs','datestamp')} function will resolve to: "20090102".

For the workflow definition with action, refer to previous example , with the following change in pig params in addition to database and table.

...
<param>PARTITION_DATE_MIN=${DATE_MIN}</param>
<param>PARTITION_DATE_MAX=${DATE_MAX}</param>
<param>REGION=${region}</param>
<param>OUT_PARTITION_VAL_REGION=${OUT_PARTITION_VAL_REGION}</param>
<param>OUT_PARTITION_VAL_DATE=${OUT_PARTITION_VAL_DATE}</param>
...

Example usage in Pig: This illustrates another pig script which filters partitions based on range, with range limits parameterized with the EL funtions

A  load '$HCAT_IN_DB.$HCAT_IN_TABLE' using org.apache.hcatalog.pig.HCatLoader();
B = FILTER A BY datestamp >
 '$PARTITION_DATE_MIN' AND datestamp < '$PARTITION_DATE_MAX' AND region=='$REGION';
C = foreach B generate foo, bar;
store C into '$HCAT_OUT_DB.$HCAT_OUT_TABLE' using org.apache.hcatalog.pig.HCatStorer('region=$OUT_PARTITION_VAL_REGION,datestamp=$OUT_PARTITION_VAL_DATE');

For example, for the 2009-01-02T00:00Z run with the given dataset instances, the above Pig script with resolved values would look like:

A  load 'myInputDatabase.myInputTable' using org.apache.hcatalog.pig.HCatLoader();
B = FILTER A BY datestamp >
 '2009010101' AND datestamp < '2009010200' AND region='APAC';
C = foreach B generate foo, bar;
store C into 'myOutputDatabase.myOutputTable' using org.apache.hcatalog.pig.HCatStorer('region=APAC,datestamp=20090102');

6.9.1. coord:dateOffset(String baseDate, int instance, String timeUnit) EL Function

The ${coord:dateOffset(String baseDate, int instance, String timeUnit)} EL function calculates date based on the following equaltion : newDate = baseDate + instance, * timeUnit

For example, if baseDate is '2009-01-01T00:00Z', instance is '2' and timeUnit is 'MONTH', the return date will be '2009-03-01T00:00Z'. If baseDate is '2009-01-01T00:00Z', instance is '1' and timeUnit is 'YEAR', the return date will be '2010-01-01T00:00Z'.

Example: :

   <coordinator-app name="app-coord" frequency="${coord:days(1)}"
                    start="2009-01-01T24:00Z" end="2009-12-31T24:00Z" timezone="UTC"
                    xmlns="uri:oozie:coordinator:0.1">
      ......
      <action>
        <workflow>
          <app-path>hdfs://bar:8020/usr/joe/logsaggretor-wf</app-path>
          <configuration>
            <property>
              <name>nextInstance</name>
              <value>${coord:dateOffset(coord:nominalTime(), 1, 'DAY')}</value>
            </property>
            <property>
             <name>previousInstance</name>
              <value>${coord:dateOffset(coord:nominalTime(), -1, 'DAY')}</value>
            </property>
         </configuration>
       </workflow>
      </action>
   </coordinator-app>

In this example, the 'nextInstance' will be '2009-01-02T24:00Z' for the first action. And the value of 'previousInstance' will be '2008-12-31T24:00Z' for the same instance.

6.9.2. coord:formatTime(String ts, String format) EL Function (since Oozie 2.3.2)

The ${coord:formatTime(String timeStamp, String format)} function allows transformation of the standard ISO8601 timestamp strings into other desired formats.

The format string should be in Java's SimpleDateFormat format.

For example, if timeStamp is '2009-01-01T00:00Z' and format is 'yyyy', the returned date string will be '2009'.

10.1.1. Dataset Names Collision Resolution

Embedded dataset definitions within a coordinator application cannot have the same name.

Dataset definitions within a dataset definition XML file cannot have the same name.

If a coordinator application includes one or more dataset definition XML files, there cannot be datasets with the same names in the 2 dataset definition XML files.

If any of the dataset name collisions occurs the coordinator job submission must fail.

If a coordinator application includes one or more dataset definition XML files and it has embedded dataset definitions, in case of dataset name collision between the included and the embedded definition files, the embedded dataset takes precedence over the included dataset.

Oozie Coordinator Schema 0.4

<xs:schema xmlns:xs="http://www.w3.org/2001/XMLSchema" xmlns:coordinator="uri:oozie:coordinator:0.2"
           elementFormDefault="qualified" targetNamespace="uri:oozie:coordinator:0.2">    <xs:element name="coordinator-app" type="coordinator:COORDINATOR-APP"/>
    <xs:element name="datasets" type="coordinator:DATASETS"/>
    <xs:simpleType name="IDENTIFIER">
        <xs:restriction base="xs:string">
            <xs:pattern value="([a-zA-Z]([\-_a-zA-Z0-9])*){1,39})"/>
        </xs:restriction>
    </xs:simpleType>
    <xs:complexType name="COORDINATOR-APP">
        <xs:sequence>
            <xs:element name="parameters" type="coordinator:PARAMETERS" minOccurs="0" maxOccurs="1"/>
            <xs:element name="controls" type="coordinator:CONTROLS" minOccurs="0" maxOccurs="1"/>
            <xs:element name="datasets" type="coordinator:DATASETS" minOccurs="0" maxOccurs="1"/>
            <xs:element name="input-events" type="coordinator:INPUTEVENTS" minOccurs="0" maxOccurs="1"/>
            <xs:element name="output-events" type="coordinator:OUTPUTEVENTS" minOccurs="0" maxOccurs="1"/>
            <xs:element name="action" type="coordinator:ACTION" minOccurs="1" maxOccurs="1"/>
        </xs:sequence>
        <xs:attribute name="name" type="coordinator:IDENTIFIER" use="required"/>
        <xs:attribute name="frequency" type="xs:string" use="required"/>
        <xs:attribute name="start" type="xs:string" use="required"/>
        <xs:attribute name="end" type="xs:string" use="required"/>
        <xs:attribute name="timezone" type="xs:string" use="required"/>
    </xs:complexType>
    <xs:complexType name="PARAMETERS">
        <xs:sequence>
            <xs:element name="property" minOccurs="1" maxOccurs="unbounded">
                <xs:complexType>
                    <xs:sequence>
                        <xs:element name="name" minOccurs="1" maxOccurs="1" type="xs:string"/>
                        <xs:element name="value" minOccurs="0" maxOccurs="1" type="xs:string"/>
                        <xs:element name="description" minOccurs="0" maxOccurs="1" type="xs:string"/>
                    </xs:sequence>
                </xs:complexType>
            </xs:element>
        </xs:sequence>
    </xs:complexType>
    <xs:complexType name="CONTROLS">
        <xs:sequence minOccurs="0" maxOccurs="1">
            <xs:element name="timeout" type="xs:string" minOccurs="0" maxOccurs="1"/>
            <xs:element name="concurrency" type="xs:string" minOccurs="0" maxOccurs="1"/>
            <xs:element name="execution" type="xs:string" minOccurs="0" maxOccurs="1"/>
            <xs:element name="throttle" type="xs:string" minOccurs="0" maxOccurs="1"/>
        </xs:sequence>
    </xs:complexType>
    <xs:complexType name="DATASETS">
        <xs:sequence minOccurs="0" maxOccurs="1">
            <xs:element name="include" type="xs:string" minOccurs="0" maxOccurs="unbounded"/>
            <xs:choice minOccurs="0" maxOccurs="unbounded">
                <xs:element name="dataset" type="coordinator:SYNCDATASET" minOccurs="0" maxOccurs="1"/>
            </xs:choice>
        </xs:sequence>
    </xs:complexType>
    <xs:complexType name="SYNCDATASET">
        <xs:sequence>
            <xs:element name="uri-template" type="xs:string" minOccurs="1" maxOccurs="1"/>
            <xs:element name="done-flag" type="xs:string" minOccurs="0" maxOccurs="1"/>
        </xs:sequence>
        <xs:attribute name="name" type="coordinator:IDENTIFIER" use="required"/>
        <xs:attribute name="frequency" type="xs:string" use="required"/>
        <xs:attribute name="initial-instance" type="xs:string" use="required"/>
        <xs:attribute name="timezone" type="xs:string" use="required"/>
    </xs:complexType>
    <xs:complexType name="INPUTEVENTS">
        <xs:sequence minOccurs="1" maxOccurs="1">
            <xs:element name="data-in" type="coordinator:DATAIN" minOccurs="1" maxOccurs="unbounded"/>
        </xs:sequence>
    </xs:complexType>
    <xs:complexType name="DATAIN">
        <xs:choice minOccurs="1" maxOccurs="1">
            <xs:element name="instance" type="xs:string" minOccurs="1" maxOccurs="unbounded"/>
            <xs:sequence minOccurs="1" maxOccurs="1">
                <xs:element name="start-instance" type="xs:string" minOccurs="1" maxOccurs="1"/>
                <xs:element name="end-instance" type="xs:string" minOccurs="1" maxOccurs="1"/>
            </xs:sequence>
        </xs:choice>
        <xs:attribute name="name" type="coordinator:IDENTIFIER" use="required"/>
        <xs:attribute name="dataset" type="coordinator:IDENTIFIER" use="required"/>
    </xs:complexType>
    <xs:complexType name="OUTPUTEVENTS">
        <xs:sequence minOccurs="1" maxOccurs="1">
            <xs:element name="data-out" type="coordinator:DATAOUT" minOccurs="1" maxOccurs="unbounded"/>
        </xs:sequence>
    </xs:complexType>
    <xs:complexType name="DATAOUT">
        <xs:sequence minOccurs="1" maxOccurs="1">
            <xs:element name="instance" type="xs:string" minOccurs="1" maxOccurs="1"/>
        </xs:sequence>
        <xs:attribute name="name" type="coordinator:IDENTIFIER" use="required"/>
        <xs:attribute name="dataset" type="coordinator:IDENTIFIER" use="required"/>
    </xs:complexType>
    <xs:complexType name="ACTION">
        <xs:sequence minOccurs="1" maxOccurs="1">
            <xs:element name="workflow" type="coordinator:WORKFLOW" minOccurs="1" maxOccurs="1"/>
            <xs:any namespace="uri:oozie:sla:0.1" minOccurs="0" maxOccurs="1"/>
        </xs:sequence>
    </xs:complexType>
    <xs:complexType name="WORKFLOW">
        <xs:sequence>
            <xs:element name="app-path" type="xs:string" minOccurs="1" maxOccurs="1"/>
            <xs:element name="configuration" type="coordinator:CONFIGURATION" minOccurs="0" maxOccurs="1"/>
        </xs:sequence>
    </xs:complexType>
    <xs:complexType name="FLAG"/>
    <xs:complexType name="CONFIGURATION">
        <xs:sequence>
            <xs:element name="property" minOccurs="1" maxOccurs="unbounded">
                <xs:complexType>
                    <xs:sequence>
                        <xs:element name="name" minOccurs="1" maxOccurs="1" type="xs:string"/>
                        <xs:element name="value" minOccurs="1" maxOccurs="1" type="xs:string"/>
                        <xs:element name="description" minOccurs="0" maxOccurs="1" type="xs:string"/>
                    </xs:sequence>
                </xs:complexType>
            </xs:element>
        </xs:sequence>
    </xs:complexType>
</xs:schema>

Oozie Coordinator Schema 0.2

<xs:schema xmlns:xs="http://www.w3.org/2001/XMLSchema" xmlns:coordinator="uri:oozie:coordinator:0.2"
           elementFormDefault="qualified" targetNamespace="uri:oozie:coordinator:0.2">    <xs:element name="coordinator-app" type="coordinator:COORDINATOR-APP"/>
    <xs:element name="datasets" type="coordinator:DATASETS"/>
    <xs:simpleType name="IDENTIFIER">
        <xs:restriction base="xs:string">
            <xs:pattern value="([a-zA-Z]([\-_a-zA-Z0-9])*){1,39})"/>
        </xs:restriction>
    </xs:simpleType>
    <xs:complexType name="COORDINATOR-APP">
        <xs:sequence>
            <xs:element name="controls" type="coordinator:CONTROLS" minOccurs="0" maxOccurs="1"/>
            <xs:element name="datasets" type="coordinator:DATASETS" minOccurs="0" maxOccurs="1"/>
            <xs:element name="input-events" type="coordinator:INPUTEVENTS" minOccurs="0" maxOccurs="1"/>
            <xs:element name="output-events" type="coordinator:OUTPUTEVENTS" minOccurs="0" maxOccurs="1"/>
            <xs:element name="action" type="coordinator:ACTION" minOccurs="1" maxOccurs="1"/>
        </xs:sequence>
        <xs:attribute name="name" type="coordinator:IDENTIFIER" use="required"/>
        <xs:attribute name="frequency" type="xs:string" use="required"/>
        <xs:attribute name="start" type="xs:string" use="required"/>
        <xs:attribute name="end" type="xs:string" use="required"/>
        <xs:attribute name="timezone" type="xs:string" use="required"/>
    </xs:complexType>
    <xs:complexType name="CONTROLS">
        <xs:sequence minOccurs="0" maxOccurs="1">
            <xs:element name="timeout" type="xs:string" minOccurs="0" maxOccurs="1"/>
            <xs:element name="concurrency" type="xs:string" minOccurs="0" maxOccurs="1"/>
            <xs:element name="execution" type="xs:string" minOccurs="0" maxOccurs="1"/>
            <xs:element name="throttle" type="xs:string" minOccurs="0" maxOccurs="1"/>
        </xs:sequence>
    </xs:complexType>
    <xs:complexType name="DATASETS">
        <xs:sequence minOccurs="0" maxOccurs="1">
            <xs:element name="include" type="xs:string" minOccurs="0" maxOccurs="unbounded"/>
            <xs:choice minOccurs="0" maxOccurs="unbounded">
                <xs:element name="dataset" type="coordinator:SYNCDATASET" minOccurs="0" maxOccurs="1"/>
            </xs:choice>
        </xs:sequence>
    </xs:complexType>
    <xs:complexType name="SYNCDATASET">
        <xs:sequence>
            <xs:element name="uri-template" type="xs:string" minOccurs="1" maxOccurs="1"/>
            <xs:element name="done-flag" type="xs:string" minOccurs="0" maxOccurs="1"/>
        </xs:sequence>
        <xs:attribute name="name" type="coordinator:IDENTIFIER" use="required"/>
        <xs:attribute name="frequency" type="xs:string" use="required"/>
        <xs:attribute name="initial-instance" type="xs:string" use="required"/>
        <xs:attribute name="timezone" type="xs:string" use="required"/>
    </xs:complexType>
    <xs:complexType name="INPUTEVENTS">
        <xs:sequence minOccurs="1" maxOccurs="1">
            <xs:element name="data-in" type="coordinator:DATAIN" minOccurs="1" maxOccurs="unbounded"/>
        </xs:sequence>
    </xs:complexType>
    <xs:complexType name="DATAIN">
        <xs:choice minOccurs="1" maxOccurs="1">
            <xs:element name="instance" type="xs:string" minOccurs="1" maxOccurs="unbounded"/>
            <xs:sequence minOccurs="1" maxOccurs="1">
                <xs:element name="start-instance" type="xs:string" minOccurs="1" maxOccurs="1"/>
                <xs:element name="end-instance" type="xs:string" minOccurs="1" maxOccurs="1"/>
            </xs:sequence>
        </xs:choice>
        <xs:attribute name="name" type="coordinator:IDENTIFIER" use="required"/>
        <xs:attribute name="dataset" type="coordinator:IDENTIFIER" use="required"/>
    </xs:complexType>
    <xs:complexType name="OUTPUTEVENTS">
        <xs:sequence minOccurs="1" maxOccurs="1">
            <xs:element name="data-out" type="coordinator:DATAOUT" minOccurs="1" maxOccurs="unbounded"/>
        </xs:sequence>
    </xs:complexType>
    <xs:complexType name="DATAOUT">
        <xs:sequence minOccurs="1" maxOccurs="1">
            <xs:element name="instance" type="xs:string" minOccurs="1" maxOccurs="1"/>
        </xs:sequence>
        <xs:attribute name="name" type="coordinator:IDENTIFIER" use="required"/>
        <xs:attribute name="dataset" type="coordinator:IDENTIFIER" use="required"/>
    </xs:complexType>
    <xs:complexType name="ACTION">
        <xs:sequence minOccurs="1" maxOccurs="1">
            <xs:element name="workflow" type="coordinator:WORKFLOW" minOccurs="1" maxOccurs="1"/>
            <xs:any namespace="uri:oozie:sla:0.1" minOccurs="0" maxOccurs="1"/>
        </xs:sequence>
    </xs:complexType>
    <xs:complexType name="WORKFLOW">
        <xs:sequence>
            <xs:element name="app-path" type="xs:string" minOccurs="1" maxOccurs="1"/>
            <xs:element name="configuration" type="coordinator:CONFIGURATION" minOccurs="0" maxOccurs="1"/>
        </xs:sequence>
    </xs:complexType>
    <xs:complexType name="FLAG"/>
    <xs:complexType name="CONFIGURATION">
        <xs:sequence>
            <xs:element name="property" minOccurs="1" maxOccurs="unbounded">
                <xs:complexType>
                    <xs:sequence>
                        <xs:element name="name" minOccurs="1" maxOccurs="1" type="xs:string"/>
                        <xs:element name="value" minOccurs="1" maxOccurs="1" type="xs:string"/>
                        <xs:element name="description" minOccurs="0" maxOccurs="1" type="xs:string"/>
                    </xs:sequence>
                </xs:complexType>
            </xs:element>
        </xs:sequence>
    </xs:complexType>
</xs:schema>

Oozie Coordinator Schema 0.1

<xs:schema xmlns:xs="http://www.w3.org/2001/XMLSchema"  xmlns:coordinator="uri:oozie:coordinator:0.1"
    elementFormDefault="qualified" targetNamespace="uri:oozie:coordinator:0.1">    <xs:element name="coordinator-app" type="coordinator:COORDINATOR-APP"/>
    <xs:element name="datasets" type="coordinator:DATASETS"/>
    <xs:simpleType name="IDENTIFIER">
        <xs:restriction base="xs:string">
            <xs:pattern value="([a-zA-Z]([\-_a-zA-Z0-9])*){1,39})"/>
        </xs:restriction>
    </xs:simpleType>
    <xs:complexType name="COORDINATOR-APP">
        <xs:sequence>
            <xs:element name="controls" type="coordinator:CONTROLS" minOccurs="0" maxOccurs="1"/>
            <xs:element name="datasets" type="coordinator:DATASETS" minOccurs="0" maxOccurs="1"/>
            <xs:element name="input-events" type="coordinator:INPUTEVENTS" minOccurs="0" maxOccurs="1"/>
            <xs:element name="output-events" type="coordinator:OUTPUTEVENTS" minOccurs="0" maxOccurs="1"/>
            <xs:element name="action" type="coordinator:ACTION" minOccurs="1" maxOccurs="1"/>
        </xs:sequence>
        <xs:attribute name="name" type="coordinator:IDENTIFIER" use="required"/>
        <xs:attribute name="frequency" type="xs:string" use="required"/>
        <xs:attribute name="start" type="xs:string" use="required"/>
        <xs:attribute name="end" type="xs:string" use="required"/>
        <xs:attribute name="timezone" type="xs:string" use="required"/>
    </xs:complexType>
    <xs:complexType name="CONTROLS">
        <xs:sequence minOccurs="0" maxOccurs="1">
            <xs:element name="timeout" type="xs:string" minOccurs="0" maxOccurs="1"/>
            <xs:element name="concurrency" type="xs:string" minOccurs="0" maxOccurs="1"/>
            <xs:element name="execution" type="xs:string" minOccurs="0" maxOccurs="1"/>
        </xs:sequence>
    </xs:complexType>
    <xs:complexType name="DATASETS">
        <xs:sequence minOccurs="0" maxOccurs="1">
            <xs:element name="include" type="xs:string" minOccurs="0" maxOccurs="unbounded"/>
            <xs:choice minOccurs="0" maxOccurs="unbounded">
                <xs:element name="dataset" type="coordinator:SYNCDATASET" minOccurs="0" maxOccurs="1"/>
            </xs:choice>
        </xs:sequence>
    </xs:complexType>
    <xs:complexType name="SYNCDATASET">
        <xs:sequence>
            <xs:element name="uri-template" type="xs:string" minOccurs="1" maxOccurs="1"/>
            <xs:element name="done-flag" type="xs:string" minOccurs="0" maxOccurs="1"/>
        </xs:sequence>
        <xs:attribute name="name" type="coordinator:IDENTIFIER" use="required"/>
        <xs:attribute name="frequency" type="xs:string" use="required"/>
        <xs:attribute name="initial-instance" type="xs:string" use="required"/>
        <xs:attribute name="timezone" type="xs:string" use="required"/>
    </xs:complexType>
    <xs:complexType name="INPUTEVENTS">
        <xs:sequence minOccurs="1" maxOccurs="1">
            <xs:element name="data-in" type="coordinator:DATAIN" minOccurs="1" maxOccurs="unbounded"/>
        </xs:sequence>
    </xs:complexType>
    <xs:complexType name="DATAIN">
        <xs:choice minOccurs="1" maxOccurs="1">
            <xs:element name="instance" type="xs:string" minOccurs="1" maxOccurs="unbounded"/>
            <xs:sequence minOccurs="1" maxOccurs="1">
                <xs:element name="start-instance" type="xs:string" minOccurs="1" maxOccurs="1"/>
                <xs:element name="end-instance" type="xs:string" minOccurs="1" maxOccurs="1"/>
            </xs:sequence>
        </xs:choice>
        <xs:attribute name="name" type="coordinator:IDENTIFIER" use="required"/>
        <xs:attribute name="dataset" type="coordinator:IDENTIFIER" use="required"/>
    </xs:complexType>
    <xs:complexType name="OUTPUTEVENTS">
        <xs:sequence minOccurs="1" maxOccurs="1">
            <xs:element name="data-out" type="coordinator:DATAOUT" minOccurs="1" maxOccurs="unbounded"/>
        </xs:sequence>
    </xs:complexType>
    <xs:complexType name="DATAOUT">
        <xs:sequence minOccurs="1" maxOccurs="1">
            <xs:element name="instance" type="xs:string" minOccurs="1" maxOccurs="1"/>
        </xs:sequence>
        <xs:attribute name="name" type="coordinator:IDENTIFIER" use="required"/>
        <xs:attribute name="dataset" type="coordinator:IDENTIFIER" use="required"/>
    </xs:complexType>
    <xs:complexType name="ACTION">
        <xs:sequence minOccurs="1" maxOccurs="1">
            <xs:element name="workflow" type="coordinator:WORKFLOW" minOccurs="1" maxOccurs="1"/>
            <xs:any namespace="uri:oozie:sla:0.1" minOccurs="0" maxOccurs="1"/>
        </xs:sequence>
    </xs:complexType>
    <xs:complexType name="WORKFLOW">
        <xs:sequence>
            <xs:element name="app-path" type="xs:string" minOccurs="1" maxOccurs="1"/>
            <xs:element name="configuration" type="coordinator:CONFIGURATION" minOccurs="0" maxOccurs="1"/>
        </xs:sequence>
    </xs:complexType>
    <xs:complexType name="FLAG"/>
    <xs:complexType name="CONFIGURATION">
        <xs:sequence>
            <xs:element name="property" minOccurs="1" maxOccurs="unbounded">
                <xs:complexType>
                    <xs:sequence>
                        <xs:element name="name" minOccurs="1" maxOccurs="1" type="xs:string"/>
                        <xs:element name="value" minOccurs="1" maxOccurs="1" type="xs:string"/>
                        <xs:element name="description" minOccurs="0" maxOccurs="1" type="xs:string"/>
                    </xs:sequence>
                </xs:complexType>
            </xs:element>
        </xs:sequence>
    </xs:complexType>
</xs:schema>

Oozie SLA Schemas

<xs:schema xmlns:xs="http://www.w3.org/2001/XMLSchema"
           xmlns:sla="uri:oozie:sla:0.2" elementFormDefault="qualified"
           targetNamespace="uri:oozie:sla:0.2">    <xs:element name="info" type="sla:SLA-INFO"/>
    <xs:complexType name="SLA-INFO">
        <xs:sequence>
            <xs:element name="nominal-time" type="xs:string" minOccurs="1"
                        maxOccurs="1"/>
            <xs:element name="should-start" type="xs:string" minOccurs="0"
                        maxOccurs="1"/>
            <xs:element name="should-end" type="xs:string" minOccurs="1"
                        maxOccurs="1"/>
            <xs:element name="max-duration" type="xs:string" minOccurs="0"
                        maxOccurs="1"/>
            <xs:element name="alert-events" type="xs:string" minOccurs="0"
                        maxOccurs="1"/>
            <xs:element name="alert-contact" type="xs:string" minOccurs="0"
                        maxOccurs="1"/>
            <xs:element name="notification-msg" type="xs:string" minOccurs="0"
                        maxOccurs="1"/>
            <xs:element name="upstream-apps" type="xs:string" minOccurs="0"
                        maxOccurs="1"/>
        </xs:sequence>
    </xs:complexType>
</xs:schema>

<?xml version="1.0" encoding="UTF-8"?>
<xs:schema xmlns:xs="http://www.w3.org/2001/XMLSchema"
	xmlns:sla="uri:oozie:sla:0.1" elementFormDefault="qualified"
	targetNamespace="uri:oozie:sla:0.1">        <xs:element name="info" type="sla:SLA-INFO" />
	<xs:complexType name="SLA-INFO">
		<xs:sequence>
			<xs:element name="app-name" type="xs:string" minOccurs="1"
				maxOccurs="1" />
			<xs:element name="nominal-time" type="xs:string"
				minOccurs="1" maxOccurs="1" />
			<xs:element name="should-start" type="xs:string"
				minOccurs="1" maxOccurs="1" />
			<xs:element name="should-end" type="xs:string" minOccurs="1"
				maxOccurs="1" />
			<xs:element name="parent-client-id" type="xs:string"
				minOccurs="0" maxOccurs="1" />
			<xs:element name="parent-sla-id" type="xs:string"
				minOccurs="0" maxOccurs="1" />
			<xs:element name="notification-msg" type="xs:string"
				minOccurs="0" maxOccurs="1" />
			<xs:element name="alert-contact" type="xs:string"
				minOccurs="1" maxOccurs="1" />
			<xs:element name="dev-contact" type="xs:string" minOccurs="1"
				maxOccurs="1" />
			<xs:element name="qa-contact" type="xs:string" minOccurs="1"
				maxOccurs="1" />
			<xs:element name="se-contact" type="xs:string" minOccurs="1"
				maxOccurs="1" />
			<xs:element name="alert-frequency" type="sla:alert-frequencyType"
				minOccurs="0" maxOccurs="1" />
			<xs:element name="alert-percentage" type="sla:alert-percentageType"
				minOccurs="0" maxOccurs="1" />
			<xs:element name="upstream-apps" type="xs:string"
				minOccurs="0" maxOccurs="1" />
		</xs:sequence>
	</xs:complexType>
    <xs:simpleType name="alert-percentageType">
         <xs:restriction base="xs:integer">
              <xs:minInclusive value="0"/>
              <xs:maxInclusive value="100"/>
         </xs:restriction>
    </xs:simpleType>
    <xs:simpleType name="alert-frequencyType">
        <xs:restriction base="xs:string">
            <xs:enumeration value="NONE"></xs:enumeration>
            <xs:enumeration value="LAST_HOUR"></xs:enumeration>
            <xs:enumeration value="LAST_DAY"></xs:enumeration>
            <xs:enumeration value="LAST_MONTH"></xs:enumeration>
        </xs:restriction>
    </xs:simpleType>
</xs:schema>