esProc External Memory Computing: Merge and Join Cursor Data

During data computing based on the table sequence, we can combine data from multiple table sequences together for use in the analysis and computation. For instance, use A.merge() to combine records of multiple table sequences in a certain order, or A.conj() to union them in order into a grand table, or JOIN functions, like join, pjoin and xjoin, to establish certain relationship between multiple table sequences by joining their data together and perform query or other computations based on this relation.

Cursors for processing big data can be merged and joined likewise. This article aims to teach the method of merging and joining data of the cursor with functions including CS.conj@x(), CS.merge@x(x) and join@x(), etc. 

1. Merge data in order

Most of the time, data could be stored in multiple data tables, for example, sales records of multiple products and employee profiles of each department. In this case, we need to merge the data from multiple data tables together for a better use. For multiple normal homogeneous table sequences, A.conj() or A.merge(x) can be used to merge the records of these table sequence into a record sequence. For big data, use CS.conj@x() and CS.merge@x(x) to combine data in every cursor of cursor sequence CS, thus the data can be fetched as a whole.

Data in the cursor will only be traversed once, so it is impossible to re-sort them after they are merged and fetched out. In view of this, the data in each cursor must be ordered in the case of merging data from multiple cursors.

Now let’s learn about the usages of and differences between CS.conj@x() and CS.merge@x(x). Firstly, let’s look at the case of simple union.

There are four text files respectively recording the order information about wines, electrical appliances, foods, and books. In A6, the four file cursors are concatenated. To find out the order in which the data are retrieved, the following code retrieves 300 records each time and data retrieval is terminated once the record of a different kind of goods appear. The result table sequence in B7 is as follows:

As can be learned from the result, regarding the unioned cursor, after all wine order data are retrieved from the 1^st text data table, the electrical appliance data from the 2^nd text data table follows. In other words, after the simple union by using CS.conj@x() function, records in the result cursor will be retrieved in the same order as each cursor is arranged in the cursor sequence CS.

In most cases, we need more than just the union of the records from each data table one after another. We want to merge them in certain order we need. To serve this purpose, CS.merge@x(x) function can be used. Please note that the records in each cursor of the cursor sequence CS must be already sorted according to expression x when using this function. For example, sort and merge the order data of various products by the sales date:

In this case, we intend to have a clear view of the order in which the records are retrieved from the cursor after they are merged in certain order. In order to do this, only the first 300 entries are retrieved. The table sequence in B7 is shown below:

It can be seen that the data are retrieved by Date in a specified order . Once all wine order data of January 1^st is retrieved, the retrieval of all electrical appliance order data of the same day will start. Because retrieving data from the cursor is a forward-only operation, the order data in each cursor must have been ordered by the date. After data of multiple cursors are merged using function CS.merge@x() in the specified order, the result data will be retrieved from the grand cursor according to the order of one of the member cursors of sequence CS, which is selected by comparing the current expression values of the data tables. In this way, we can ultimately get the result arranged in the specified order. In data retrieving, each cursor will still traverse the records in each data table for once.

Merging the data in multiple cursors in a certain order simply creates a single grand cursor, and adjusts the orders in which data are retrieved from each cursor, without increasing or decreasing any records.

If the cursor data are not ordered, they must be sorted before merging, as illustrated in the following example:

Before cursors are merged in proper order by the product number, we must ensure the data in each cursor are ordered against the product number. To do so, A5 uses cs.sortx() function to sort cursors of various products.

Please note that the cursor and the table sequence are sorted differently. Because there is usually a great amount of data in the cursor, they cannot be loaded into the memory all at once for sorting. Therefore, the data retrieval will be performed along with the data sorting. The sorted data will be saved as a temporary data file when they are accumulated to a certain amount. Once all data are retrieved and sorted, all temporary data files will be merged in proper order and returned as the cursor. For more details about sorting cursor data, please see esProc External MemorySorting: Principle of Sorting.

In B7, the retrieved records are shown below:

It can be seen that the ordered merging is accomplished once the data in each cursor have been sorted.

2. Aligned joining

When making the statistics, sometimes we need to consolidate the data from multiple cursors, which is similar to joining the data from multiple tables together. If the data in cursor need to point to a normal table sequence, then we can use cs.switch().

What if the data to be joined are all from the cursor? As we know, it is usually impossible to retrieve all data from the cursor. How can we join these data then? In esProc, we can use join@x() function to join the data from multiple cursors together. For example:

In A5~A8, perform aggregate operations over each kind of product, and return the cursor of respective temporary files (Usage of groupx is explained in detail in esProc ExternalMemory Computing: Principle of Grouping). In A9, the daily sales data of each kind of products will be aligned and joined by the date. A10 fetches the statistical results of the first 25 days, as shown below:

After cursors are joined in alignment, a grand cursor will be returned, from which the fetched result is similar to the result of joining table sequences together, that is, every field in every row is composed of the record. Thus we must note that the joined data take up more memory than those in the normal cases. In addition, since data are composed of records instead of values, please note the writing form of the expression, particularly when re-joining the result cursors used for computation.

The result of aligned joining of cursors can also be reused for filtering or generating data, for example:

From the joined cursor, A10 selects records in which the total amount of food orders is greater than that of wine orders. Then A11 generates the result cursor. A12 returns the first 100 records:

In performing the aligned joining of cursors, we must remember that the data in the cursor cannot be fetched into and store in the memory. They can only be traversed once from the first to the last. Therefore the data in each cursor must be sorted for aligned joining, which is different from handling the database multi-table join and the join() operation for normal table sequences. As shown in the above example, the data in A5~A8 are already sorted by date to ensure a correct computation.

In order to better explain this point, we create a cursor using two in-memory table sequences in which data are not big:

The table sequences in A1 and A2 are shown below:

In A5, we will see the aligned joining result:

The cursor data are different to those in the normal table sequence. During the operation of aligned joining, when looking for the New York state corresponding to the New York city, the cursor for the State data has already moved to entry 32 and the records before it are unavailable for later computations. So most cities cannot find their corresponding states. Furthermore, options @1 and @a are not used in function join@x() to specify the left join or full join. Only data of few cities finding out corresponding states are returned.

But if having the city data sorted first, we can have the normal result:

In A1, the data are sorted by STATEID:

In A6, we can see the joining result:

The joining of data from multiple cursors doesn’t require the same cursor type. For example:

A5 joins personnel data with the corresponding state data. What’s important is that the cursor data have been sorted. A6 fetches the first 100 rows of data from A5’s cursor:

In fact the aim of joining data from different cursors is usually to get the related detail information from them. In the above example, there’s no need to join records from the two cursors together if the purpose is only to get the full names of the states where employees come from. Because the state data actually come from a table sequence, cs.switch() can be used to convert a certain field of cursor cs into the corresponding records of the table sequence. There are explanations about the function’s usage in esProc External Memory Computing: BasicUsages of the Cursor. Difference between join@x() and cs.switch() is that the former creates a mutual reference between different cursors, whereas the latter establishes a relationship, through the foreign key, between one or more certain fields of a cursor and a table sequence/sequence, which is regarded as a dimension table in the process. join@x() for joining cursor data requires the data be sorted by the field according to which they are joined. cs.switch() hasn’t such a requirement as the dimension table in use can be referenced directly from the memory.

Sometimes, in a foreign-key relationship, the composite foreign key is needed to reference certain data of a table sequence or a cursor. In this case cs.join() can be used to append fields to the cursor. For example: 

A4 creates a table sequence based on the month and the states’ abbreviations and A5 inserts detailed data into it. Then A4’s table sequence is as follows:

A6 appends fields to A2’s cursor using cs.join(). Equally, the Month field holding employees’ birthday information and the State field of A2’s cursor are joined with the Month and State field of A4’s cursor. Then fetch GroupID field from A4’s cursor and append it to State field of A2’s cursor after the joining. A7 fetches the first 100 records from A6’s cursor:

With cs.join(), we can join not only the cursor, but the table sequence, with the field in another cursor. Different from join@x(),cs.join() doesn’t require data in the cursor cs be sorted. But if the dimension table in use comes from the cursor, its data must be fetched into the memory all at one. With the requirement of joining two fields of the cursor cs and the dimension table, this example cannot be handled with cs.switch(). 

3 Unconditional joining

In the example in the previous section, the statistical data of orders of different products are joined by the dates. In reality the number of rows of data of every kind of order is always the same and so is the data’s order in terms of the date. Thus we don’t have to specify a criterion for sorting. We simply join the data together using pjoin function, without any condition.

A9 joins cursors of sales data of four products together, without specified order. A10 fetches the first 100 records:

We can see from the result the way fields from different cursors are joined. The result can be rearranged as follows:

A11 fetches the first 100 records:

esProc External Memory Computing: Aggregate Operations with Cursor

The data volume of the big data table is usually quite huge, which makes it impossible to retrieve all data from the big data table at once. In view of this, the data processing over big data table is usually to serve two purposes: With cs.fetch(), retrieve partial data each time or group & aggregate the data in the big data table.

1.Group and aggregate data of a cursor

The commonest grouping & aggregate operations are count and summation. In addition, the aggregate operation includes computing the maximum/minimum value and top-n queries. For the aggregate operation over the cursor, you can use function cs.groups(). The aggregate functions corresponding to various aggregate operations include count/sum/max/min/topx. For example:

In A7, make statistics on the total number of orders, maximum trade amount, and total amount of different kinds of products. One thing to note is that the precision of the double-precision data is usually not enough. In this case, decimal() is used to convert the data into the big decimal type for computing. Regarding the grouping and aggregate computation with the cursor, all data will be traversed just once. During the traversal, each row of data will be aggregated based on the group to which it belongs. Multiple aggregate values can be computed in one traversal, and the cursor will be closed automatically once the traversal is completed.

The result in A7 is as follows:

As can be seen from the result, in the text data for test, each product has 50,000 records of orders. Data of every file and the data of A6’s cursor are sorted by Date, but not sorted by Type. Seen from the use of cs.groups() function in the example, the function groups and summarizes data in the cursor without requiring them to have been sorted. After data grouping and aggregation, result will be sorted by the grouping value. In addition, because cs.groups() returns a table sequence, the result set must not exceed the memory size when the function is used to handle the grouping and aggregate operations on the cursor. Thus it is not fit for the cases in which the result set holds big data.

To find the n greatest/smallest data values is also one of the aggregate operations. In this case, use the aggregate function topx:

A7 computes the bottom 3 orders with the lowest total amount and the top 3 orders with the highest total amount among the orders of each kind of product. Please remember to add the negative sign to the aggregate function when computing the top 3 orders, that is, topx(-Amount:3). The result is as follows:

To aggregate all records, simply omit the grouping expression from cs.groups function:

The result in A7 is as follows:

This shows that the aggregation of all records can be regarded as a special kind of grouping and aggregate operation.

2.Aggregate data from multiple cursors

When handling the aggregate operations over multiple cursors, you can compute the aggregate result of each cursor first:

A6 groups and aggregates each file cursor to compute the daily aggregated statistics of each kind of product:

Because the data will be sorted by grouping value when using the cs.groups() function to aggregate and group the cursor data, the result table sequence of aggregating the order data of each kind of product is always ordered by the date, whatever the order of original data is. So, the result can be merged further to compute the daily order data for all products:

In A8, you will see the summarizing result:

Since the order data of each kind of product in every data file have been sorted by date, cursors can be merged first, and then aggregated:

As you can see, this method leads to the same result as that got in the above. For cursors each of which data are ordered, it would be simpler and more effective to merge first, then group and aggregate. Note the difference between merge and merge@x. A.merge() is to merge the members of multiple sequences or records of multiple table sequences in proper order, and return a sequence. The computation is completed during the execution; CS.merge@x() is to merge the records from multiple cursors in proper order, return a cursor, and start the computing when fetching data from the cursor. Both functions require that the data in A and CS are ordered. There are detailed explanations about the usage of merge@x in esProc External Memory Computing: Merge and Join Cursor Data.

3.Big result sets   

In the above section, we’ve learned that the data of each cursor can be grouped and aggregated first to have a group of ordered result table sequences, which then will be merged to get the final result. With such method, we can solve the grouping and aggregating problems related to the big result sets.

The big result set means not only the source data for computation is of huge amount, but the computed result has huge data too. Owing to this, the computed result set cannot be read into the memory all at once. It has to be read out step by step. For example, the monthly bill for each customer of telecommunication company and the sales of each product of the B2C website. These statistical results may contain more than several millions of records.

In esProc, you can use cs.groupx() function to perform the grouping and aggregate operations over the big result set. Here, we will take the daily statistic on product orders as the example to illustrate the processing of big result sets. In order to have a better understanding about the memory limit, we retrieve 100 records each time. Thus, if we need to compute the daily order by grouping and aggregating, we should use groupx function:

In A6, the data in each cursor are concatenated in proper order, instead of being sorted by date. A7 returns the result of the grouping and aggregate operation as a cursor instead of a table sequence. By doing so, the data can be retrieved step by step in A8~A11. The data are shown below:

In which, each of A8~A10 retrieves the statistical result for 100 days, and A11 retrieves the remaining data. After the data of A7’s cursor are all fetched out, the cursor will close automatically.

In A7, with groupx function, the data are grouped and aggregated by date, and the number of buffer rows is set to 100. By doing so, when executed, A7 will fetch data from A6’s cursor and meanwhile, split and aggregate them. Once the result of aggregating 100 rows is produced, it will be buffered to a temporary file, and the rest can be done in the same manner. The computed result in A7 is a cursor composed of these temporary files:

Through step-by-step execution, you can view the generated temporary files in the system temporary directory after A7 is executed:

In order to see the contents of each of these binary files, you can retrieve the data from them, for example:

The data in A1 and A3 are shown below:

The name of these temporary files are generated randomly. Judging from the data retrieved from some temporary files, each temporary file stores the grouping and aggregate result of some consecutive original records, which are already sorted by the date. In fact, based on the number of buffer rows specified in the function, each temporary file, except for the last one, contains partial summarizing data for just 100 days. For a cursor generated with groupx, its data will be fetched with fetch by merging the data of all temporary files in proper order.

After cursor data are fetched out completelyby using the temporary file cursors or these cursors are closed, the corresponding temporary files will be deleted automatically. About the external memory grouping, please refer to esProc External Memory Computing:Principle of Grouping.

esProc External Memory Computing: Basic Usages of the Cursor

esProc supports importing big data in batches with the cursor, which is the usual method used in big data computing. Usages of cursors, including external file cursor, database cursor and in-memory record sequence cursor, are basically the same. This article will take the external file cursor as the example to explain basic usages of the cursor in esProc.

1. Data fetching

When a cursor is created, its data can be fetched by cs.fecth() function. Different from the way data of an ordinary table are imported, data in the cursor are fetched by specifying the number of rows to be returned, or one or more conditions on which data will be fetched, rather than being fetched all at once. For example:

A2 creates an external file cursor. A3 fetches the first 1,000 rows from it, as shown below:

With fetch function, each batch of data fetched from the cursor will be returned as a table sequence. A4 calls cs.close() function to close the cursor. When a cursor is closed, data cannot be fetched from it. That is why the result of A5 is null:

When the cursor has accomplished its mission and is not needed any more, it ought to be closed through calling cs.close() function to release the memory.

Besides fetching data from the file or the database result set with the cursor, cs.skip() function can be used to skip the specified number of rows in the cursor. For example:

A2 creates the cursor and A3 skips 45,000 rows. Then A4 fetches data from the cursor as follows:

As can be seen from the transaction ID, the data fetching begins from the 45001th row. Notice that the fetch function in A4 doesn’t specify the number of rows to be returned. In this case, all the rest of the data in the cursor will be fetched. Once data in the cursor are completely fetched, the cursor closes automatically, without the necessity of calling cs.close() function.

So A5 fetches nothing from A2’s cursor:

In fact, both cs.skip() function and cs.fecth() function will only traverse the cursor data once from front to back. When the data fetching is over or cs.close() function is executed, the cursor becomes useless.

Another type of fetching data from the cursor using cs.fecth() function is cs.fetch(;x), which specifies the condition on which data fetching will continue interruptedly until the value of expression x changes. For example:

A2 creates the cursor. A3 fetches data from it until the value of Date changes. The fetched data are as follows:

It can be seen that A3 fetches all data whose Date value is 2013-01-01. When A4 goes on with fetching data from A2’s cursor, it will begin from the date 2013-01-02:

Note that we cannot specify both the number of rows to be returned and the condition for data fetching at the same time; otherwise only the former is valid. Besides, since not all the data in A2’s cursor have been fetched, the cursor needs to be closed in A5 after doing its job.

Similarly, we can also specify condition for cs.skip() function to skip certain records. For example:

A2 creates the cursor. A3, A4 and A5 respectively skip records of one day. Then A6 will fetch data from the cursor from the fourth day:

Similarly, if not all the data have been fetched from the cursor, call cs.close() function in A7 to close the cursor.

2. Cursor loop

When fetching data from the cursor, use for statement to process the rows returned from the cursor by loop. For example:

A2 creates the cursor. By loop, A3 processes 1,000 records returned by the cursor each time. After all records are traversed, the cursor closes automatically. In A3’s code block, based on the table sequence returned by the cursor each time, the code computes the total sales amount of the salesperson whose ID number is 1 and stores the result in B2. The value is shown as follows:

Sometimes not all data need to be traversed during the cursor loop. For example:

A2 creates the cursor and A3 executes cursor loop, fetching 1,000 records each time. In A3’s code block, according to the batches of data fetched from the cursor, the code selects the sale records of the salesperson whose ID number is 1 and stores them in B2. Exit the loop after the first 5 records are fetched. At this point the result seen in B2 is as follows:

One point worthy of special note: When the computation is over and the loop is terminated by break function, there are data still remaining in the cursor. Different from the previous example, in this case the cursor won’t close automatically, thus close function is called in A6 to close it and release the memory.

Similar to cs.fecth() and cs.skip(), the condition can be specified for the cursor loop, so that the data fetching will stop as soon as the value of the conditional expression changes. For example:

With the condition, A3 fetches the sales data of one day each time, computes the number of transactions and the total sales amount in this day in its code block and stores the results in B2’s table sequence. After the code is executed, B2’s result table sequence is as follows:

3. Computation with the cursor

Besides data fetching, esProc offers the use of cursor in performing common forms of data computing by directly using certain functions. The following example is to select data from the cursor according to the condition:

B1 creates the external file cursor and A2 selects the sales data of the salesperson whose ID number is 1 from B1’s cursor. A2’s result is as follows:

It can be seen that, different from the table sequence or the record sequence, the select function used with the cursor returns the cursor as the result. Actually A2 hasn’t really got data from the cursor. It is in B2 where cs.fecth() function is being executed that the eligible data are fetched from B1’s cursor according to the condition. Result of B2 is as follows:

No number of rows to be returned or the condition for data fetching is specified for cs.fecth() function used in B2. All data, or all the sales data of the salesperson, in A2’s cursor will be returned. After data of the cursor in A2 have all been fetched out, the cursor will close automatically. Since the cursor in A2 is generated from the one in B1, the data query is in fact executed by traversing data of the cursor in B1 and the completion of fetching data from the cursor in A2 means the same thing to the cursor in B1, which will close at the same time. Therefore the effort of A3 and B3 trying to fetch data from the cursors in either A2 or B1 will definitely fail.

The cursor can also be used to create new records, or append new fields. For example:

B1 creates a file cursor using some of the fields of the external file employee.txt. In A2, the employees’ full names are generated by joining NAME field and SURNAME field in B1’s cursor together. B2 computes each employee’s age according to the BIRTHDAY field in B1’s cursor and insert the results to Age field. Similar to cs.select(), both cs.new() and cs.derive() return cursor as the result, rather than returning data directly. When executed, A2 and B2 have the following results:

A3 fetches the first 100 rows from A2’s cursor:

Generating data with cs.new() function means executing new statement on every record.

B3 fetches the first 100 rows from B2’s cursor:

Generating data with cs.derive() function means appending computed columns into the returned table sequence.

It can be seen that B3’s data begin not from the first employee, but from the 101th one. That’s because fetching data from cursors in A2 and B2 is, in reality, fetching data from the cursor in B1 and cursor data will be traversed only once from front to back. After A3 is executed, the first 100 rows of data in B1’s cursor have been traversed. So B3 can only go on with the traversal in order in the execution of cs.fecth() function by beginning with the 101th employee.

Because data haven’t been all fetched out, the cursors need to be closed using cs.close() function. As A4 closes the cursor in A2, the cursor in B1 is also closed simultaneously, for the former is generated by the latter. This affects the cursor in B2, from which B4 is trying to fetch data but to no avail.

In practical use, we ought to avoid using the same cursor to generate multiple ones as possibly as we can; the same problem as the one with B3 in fetching data, or as another one that cursors will be unavailable when the source cursor closes, will arise otherwise.

What’s more, cs.run() can be used to modify values of a certain field of cursor data. For example:

B2 replaces names of the states in the returned data by their abbreviations and still returns the cursor as the result:

A3 fetches the first 100 rows as follows:

Similar to a table sequence, the cursor allows the conversion of a certain field of its data into the corresponding records in another table sequence by using cs.switch() function. For example:

B2 uses cs.switch() to switch certain data of the cursor. Its result is also a cursor:

The data A3 fetches are as follows：

It can be seen that, in the return result, STATE field has been converted into the corresponding state records.

cs.switch() can convert more than one field of the cursor. For example:

B3 switches records of several fields of the cursor to the corresponding ones in another table sequence. A4 fetches the first 100 records:

In the article – RelatedComputing in esProc – Primary Keys and Index Function, we explained the usage of switch in the ordinary table sequence, which is similar to that of cs.switch() in external memory computing. Both functions will create corresponding index table for certain field(s) of the dimension table in order to increase efficiency. It is more important to adopt this processing method while using the cursor because the data under processing are usually big.