Dataset

A dataset is a logical entity to represent a set of data produced by an application. A user can define a dataset either using its directory location or using metadata. Oozie has always supported directory-based data dependency. Recently, Oozie introduced metadata-based data dependency as well. This book primarily focuses on the directory-based dataset, as that’s the most commonly used approach. The metadata-based dependency will be covered later, in “HCatalog-Based Data Dependency”.

Furthermore, the data in a dataset can be produced in two ways:

• In a fixed interval

• Ad hoc/random, without following any time pattern

In Oozie, the dataset produced in regular frequency is called synchronous and the dataset produced randomly is known as asynchronous. Oozie currently supports only the synchronous datasets, so this is the type we will focus on in this chapter. We will explore an approach to handle asynchronous datasets later in “Emulate Asynchronous Data Processing”. In a nutshell, a dataset in Oozie is a template to represent a set of directory-based data produced at fixed time intervals.

<dataset name="ds_input1" frequency="${coord:hours(6)}" 
      initial-instance="2014-12-29T02:00Z">
  <uri-template> 
     ${baseDataDir}/revenue_feed/${YEAR}-${MONTH}-${DAY}-${HOUR}
   </uri-template>
   <done-flag>_trigger</done-flag>
</dataset>

<datasets>
  <include>hdfs://localhost:8020/user/joe/shares/common_datasets.xml</include>
  <dataset name="ds_input1" frequency="${coord:hours(6)}" 
       initial-instance="2014-12-29T02:00Z">
    <uri-template> 
      ${baseDataDir}/revenue_feed/${YEAR}-${MONTH}-${DAY}-${HOUR}
    </uri-template>
    <done-flag>_trigger</done-flag>
  </dataset>
</datasets>

<datasets>
  <dataset name="ds_input2" frequency="${coord:hours(6)}"
       initial-instance="2014-12-29T02:00Z">
    <uri-template> 
      ${baseDataDir}/revenue_feed/${YEAR}-${MONTH}-${DAY}-${HOUR}
    </uri-template>
    <done-flag>_trigger</done-flag>
  </dataset>
</datasets>

oozie.service.HadoopAccessorService.
  supported.filesystems

<input-events>
  <data-in name="event_input1" dataset="ds_input1">
    <start-instance>${coord:current(-4)}</start-instance>
    <end-instance>${coord:current(-1)}</end-instance>
  </data-in>
</input-events>

<output-events>
  <data-out name="event_output1" dataset="daily-feed">
    <instance>${coord:current(0)}</instance>
  </data-out>
</output-events>

current(n)

This EL function returns the timestamp of the n^th instance of a dataset relative to a specific coordinator action’s nominal time (caNT). The value of n can be any integer number. Any negative value for n refers to an instance earlier than the nominal time. While any positive value for n refers to some instance after the nominal time. The simplest equation to approximately calculate the timestamp of the n^th instance is as follows:

    current(n) = dsII + dsF * (n + (caNT – dsII) / dsF)

The following example further clarifies the concept with real values. Assume the coordinator job has the start time of 2014-10-18T06:00Z and a frequency of one day. This means Oozie will materialize coordinator actions with the following nominal times (in order): 2014-10-18T06:00Z, 2014-10-19T06:00Z, 2014-10-20T06:00Z, and so on. Let’s also assume that there are four datasets with the attributes in Table 7-2 to demonstrate the different scenarios.

We explain below how to calculate some of the time instances of these datasets. This calculation is in the context of the second coordinator action with a nominal time of 2014-10-19T06:00Z.

current(0) of ds1: The current(0) of any dataset specifies the dataset instance/timestamp that is closest to and no later than the coordinator action’s nominal time. In general, finding current(0) is the first step in understanding any other data instance. Most instance calculations are based on the coordinator action’s nominal time. Conceptually, we can start from the dataset’s initial instance and go forward to the coordinator action’s nominal time with an increment of dataset frequency. In this example, we start with dsII=2014-10-06T06:00Z and go toward caNT= 2014-10-19T06:00Z with a frequency of one day. The dataset instances will correspond to 10/6, 10/7, 10/8, and so on (in order). In this example, the nominal time and the dataset (ds1) initial timestamp have the same time component (6 a.m.) and that makes the calculation a little easier. So the closest dataset timestamp is the same as the nominal time and is 2014-10-19T06:00Z. Hence this time represents current(0) as well. We can also calculate the same using the following equation:

current(n) = dsII + dsF * (n + (caNT – dsII) / dsF)
           = 2014-10-06T06:00Z + 1 day x 
               (0 + (2014-10-19T06:00Z - 2014-10-06T06:00Z))/ 1 day
           = 2014-10-06T06:00Z + 13 day = 2014-10-19T06:00Z

Similarly, we can calculate current(-1), which is the immediate previous instance of current(0), and current(1), which is the immediate next instance of current(0). We also describe the same concept in Figure 7-2.

Figure 7-2. Timestamps of current() EL function for dataset ds1

current(0) of ds3: We use dataset ds3 to explain the same idea in a slightly different scenario. In this example, the dataset instances starts with dsII=2014-10-06T06:00Z and moves toward caNT= 2014-10-19T06:00Z with a frequency of three days. Dataset instances will be 10/6, 10/9, 10/12, 10/15, 10/18, 10/21, and so on (in order). So the closest instance to nominal time is 10/18, which becomes current(0) for this scenario. Notably, the nominal time 2014-10-19T06:00Z and current(0) do not exactly match in this example. Figure 7-3 displays the different data instances including current(-1), current(1), and so on.

Figure 7-3. Timestamps of current() EL function for dataset ds3

Table 7-3 shows the return value of current(n) given different values of n for all of the example datasets in Table 7-2.

Instances before the dataset’s initial-instance: The data instances before the initial-instance of any dataset doesn’t count. So if the EL function (e.g., current()) refers to any such dataset instance, coordinator doesn’t really check the existence of that data. In other words, there could be some data on HDFS before the dataset’s initial-instance as defined in the dataset definition, but Oozie disregards those data instances. However, Oozie returns an empty ("") string for any such instance. For instance, current(-14) for dataset ds1 points to 2014-10-05T06:00Z, which is earlier than the declared initial-instance (2014-10-06T06:00Z) of ds1. In this case, Oozie returns an empty string ("") without checking for the existence of the data.

Scope: The current() EL function is valid only within the <data-in> and <data-out> sections of a coordinator XML.

latest(n)

This EL function returns the timestamp of the n^th latest available dataset. Evaluating this “latest” available data instance happens with respect to either one of the two points in time listed here:

present time

The wall-clock time when Oozie evaluates the latest() function

actual time

The time when Oozie actually materializes a coordinator action

Oozie selects this option based on the property oozie.service.ELService.latest-el.use-current-time defined in the oozie-site.xml file. The default is to evaluate “latest” based on the action’s actual time.

Nominal versus actual versus present time

Before going further into the explanation of the latest() function, we need to clarify the newly introduced terms related to time. We are already familiar with the action nominal time. We just introduced two new terms, present time and actual time. Present time represents the current wall clock time when the latest evaluation logic is executed. In other words, if the same latest function is executed multiple times, it will obviously use different present (wall-clock) times for its dataset evaluations.

On the other hand, the action’s actual time represents the time when the action is materialized by Oozie. Although this sounds very similar to nominal time, there are subtle but important differences. For instance, when a coordinator is delayed and is running in catch-up mode, an action may be actually created at 6 p.m. but it should have been ideally created at 2 p.m. In other words, action’s nominal time is 2 p.m., but the action’s actual time is 6 p.m.

Now let’s also assume this delayed action has a latest() dependency and is checking and waiting for data availability. For example, at 10 p.m., the coordinator evaluates the latest() function. At that moment, the present time on the wall clock is 10 p.m. whereas the nominal time (2 p.m.) and the actual time (6 p.m.) remain unchanged. In short, the nominal time of an action is always fixed, the actual time becomes fixed once the action is created, and the present time is always changing and follows the wall clock.

latest() evaluation

As already mentioned, Oozie evaluates latest(n) based on either the coordinator action’s actual time or the present wall clock time. Let’s generalize this time as look-back start time denoted by Tlbs. At first, Oozie determines the closest time instance of the dataset to Tlbs. Oozie starts from the dataset’s initial-instance (dsII) and increments it by the dataset’s frequency (dsF) until it reaches Tlbs. Let’s assume the closest timestamp value to Tlbs is determined to be Tds. Oozie first checks if the data directory for time Tds is available. If it is available, it will consider it as the first available data instance or latest(0). If the data for Tds is not available yet, Oozie will walk back and look for data for time Tds - dsF (dataset frequency). If that data is available, Oozie will consider this second time instance as the first available data instance or latest(0). If data is not available for that instance as well, Oozie will skip it and keep walking back.

Continuing with this example, if all previous data instances are available, the n^th available instance (latest(n)) will be the data instance for time Tds - n x dsF. If any one data instance between time Tds and Tds - n x dsF is not available for whatever reason, Oozie needs to look back further to find (latest(n). If it can’t get to the n^th instance after searching backward all the way to the initial-instance of the dataset (dsII), Oozie will go to sleep and start the evaluation process again in the next cycle starting with the calculation of the time Tlbs. Finally, when Oozie finds the n^th available instance, it returns the corresponding timestamp.

The following detailed example attempts to further clarify the concept with real values. Assume the coordinator job specifies the start time as 2014-10-18T06:00Z and the frequency as one day. Oozie materializes the first coordinator action with a nominal time of 2014-10-18T06:00Z.

Let’s further assume that the dependent dataset has the following attributes:

• Initial-instance = 2014-10-06T06:00Z

• frequency = 1 day

• uri-template = hdfs://foo:8020/logs/${YEAR}-${MONTH}-${DAY}

Let’s consider the scenario where Oozie is evaluating latest()for this dataset at two different times on the wall clock: 2014-10-19T10:00Z and 2014-10-19T11:00Z. At these times, let’s also assume that the actual data availability is as it appears in Table 7-4.

Table 7-5 shows the return value of the latest(n) timestamp for various values of n given the above example scenario. The example assumes that the property oozie.service.ELService.latest-el.use-current-time is set to true. In other words, it utilizes the present wall-clock time (instead of the actual time) in evaluating latest().

latest() at 10 a.m.: At first, Oozie determines the closest timestamp that could be the candidate for latest(0). It starts from 2014-10-06T06:00Z (dataset initial-instance, dsII) and increments the timestamp using the frequency (dsF) until it reaches the present time (2014-10-19T10:00Z). In this example, that instance evaluates to 2014-10-19T06:00Z and that’s where Oozie starts its data availability checks. Since the data from 10/19 is available at wall-clock time 10 a.m., that instance is determined to be the latest(0). But for latest(-1), Oozie looks for the 10/18 data, which is actually missing at wall-clock time 10 a.m. So Oozie continues to look backward and finds there is no data for 10/17 either. However, it finds data for 10/16 and returns that as the latest(-1). Using the same approach, it looks backwards for latest(-2) and skips 10/15 due to missing data. Oozie finally finds data in 10/14 and returns that instance as the latest(-2). We demonstrate this pictorially in Figure 7-4.

Figure 7-4. Timestamps of latest() EL function at wall-clock time 10 a.m.

latest() at 11 a.m.: The only context change between wall clock time 10 a.m. and 11 a.m. is the arrival of new data for 10/18. This changes the timestamp evaluations for latest(-1) and latest(-2). Specifically, if Oozie evaluates at 11 a.m., it will return 10/18 as latest(-1) and 10/16 as latest(-2).

Scope: The latest() EL function is valid only within the <data-in> and <data-out> sections of the coordinator XML.

dataIn(eventName):

This function evaluates all input data directories of a dataset for a specific time instance and returns the directories as a string. The dataIn() function doesn’t really check if the data is available or not. This function takes eventName as a parameter. First, Oozie identifies data-in from the input-events definition using the eventName. Second, Oozie finds the name of the dataset from the data-in definition. Last, Oozie takes the uri-template from the dataset definition and resolves the paths corresponding to the particular time instance. Oozie evaluates the time instance based on nominal time and the instance number defined in the EL function. We already saw the details of this process in “Parameterization of Dataset Instances”. If there are multiple instances (e.g., current(0), current(-1), etc.) in data-in, Oozie concatenates them using , as a separator.

For instance, consider the EL function ${coord:dataIn('event_input1')} in the context of the example dataset and input-events defined in “Defining a dataset” and “input-events”, respectively. Let’s also assume the nominal time of the coordinator action is 2015-01-01T02:00Z. Using the event name event_input1, Oozie initially determines the corresponding dataset name ds_input1 from the data-in definition. Then using the nominal time and instance count (such as -1 for current(-1)), Oozie calculates the exact time instance of the data and ultimately resolves the uri-template defined in the dataset. For example, current(-1) returns 2014-12-31T20:00Z for this coordinator action. Finally, Oozie resolves the uri-template with this time instance and evaluates the final directory as hdfs://localhost:8020/user/joe/revenue_feed/2013-12-31-20. Oozie follows the same process for each current instance and concatenates them with a comma (,). The EL function dataIn() finally returns the following string:

        
      hdfs://localhost:8020/user/joe/revenue_feed/2014-12-31-02,
      hdfs://localhost:8020/user/joe/revenue_feed/2014-12-31-08,
      hdfs://localhost:8020/user/joe/revenue_feed/2014-12-31-14,
      hdfs://localhost:8020/user/joe/revenue_feed/2014-12-31-20
      
    

Scope: dataIn() is valid within the <workflow> section of a coordinator XML.

dataOut(eventName)

This function is similar to dataIn(). The key difference is that dataOut() utilizes the output-events and data-out sections whereas the dataIn() uses input-events and data-in .

Scope: dataOut() is valid within the <workflow> section of a coordinator XML.

nominalTime()

This function returns the nominal time or the action creation time (explained in section “Our First Coordinator Job”) of a particular coordinator action.

Scope: nominalTime() is valid within the <workflow> section of a coordinator XML.

actualTime()

This function calculates the actual time of a coordinator action as defined in “Parameterization of Dataset Instances”. In an ideal world, the nominal time and the actual time of an action will be the same. But during catch-up scenarios, where the coordinator action execution is delayed, the actual time of a coordinator action is different and later than its nominal time.

Scope: actualTime() is valid within the <workflow> section of a coordinator XML.

dateOffset(baseTimeStamp, skipInstance, timeUnit)

This utility function returns a date as a string using the base time and offset. Oozie calculates the new date using the following equation (not considering Daylight Saving Time).

New Date = baseTimeStamp + skipInstance * timeUnit

Scope: dateOffset() is valid within the <workflow> section of a coordinator XML.

formatTime(timeStamp, formatString)

This utility function formats a standard ISO8601 compliant timestamp into another timestamp string based on formatString. The formatString should follow the conventions used in Java’s SimpleDateFormat .

Scope: formatTime() is valid within the <workflow> and <input-events> sections of a coordinator XML.