A core component of each data warehouse architecture is the Extract-Transform-Load (ETL) layer. The processes in this layer integrate, transform, and move large volumes of data from data sources into a data warehouse (or data lake). For this reason, eficient execution of ETL processes is of high practical importance. Research approaches propose task re-ordering and parallel processing. In practice, companies apply scale-up or scale-out techniques to increase performance of the ETL layer. ETL tools existing on the market support mainly parallel processing of ETL tasks. Additionally, the IBM DataStage ETL engine applies the so-called balanced optimization that allows to execute ETL tasks either in a data source or in a data warehouse, thus optimizing overall resource consumption. With the widespread of non-relational technologies for storing big data, it is essential to apply ETL processes to this type of data. In this paper, we present an initial assessment of the balanced optimization applied to IBM DataStage, ingesting data from Kafka streaming data source.
A core component of a data warehouse (DW) [25] or a data lake
(DL) [22] is an Extract-Transform-Load (ETL) software. It is
responsible for ingesting data from data sources (DSs), integrating,
and cleaning data as well as uploading them into a DW/DL. ETL
is implemented as a workflow (process) composed of tasks,
managed by an ETL engine, frequently run on a dedicated server or
on a cluster. An ETL process moves large volumes of data
between DSs and a DW. Therefore, its execution may take hours to
complete. For example, simple movement of 1TB of data between
a DS and DW (which use magnetic disks with a typical 200MB/s
throughput) takes 2.7 hours by an ETL process. Resource
consumption and the overall processing time are further increased
by complex tasks executed within the process, including
integrating, cleaning, and de-duplicating data. For this reason, providing
means for optimization of ETL processes is of high practical
importance. Big data add to the problem more complexity that
results from: (
Unfortunately, this problem has been only partially addressed by technology and research, cf. Section 2. In practice, companies apply scale-up or scale-out for their ETL servers, to reduce processing time. ETL tools existing on the market support mainly parallel processing of ETL tasks. Some of them, on top of that, apply task movement within an ETL process. One of such techniques is called balanced optimization [17]. Research approaches propose task re-ordering and parallel processing.
This paper presents first findings from an R&D project, jointly
done by IBM Software Lab and Poznan University of
Technology, on assessing the performance of the balanced optimization
on a streaming data source. In particular, we were interested in
assessing: (
In this section we overview the solutions to optimizing ETL
processes: (
Industrial projects utilizing ETL, reduce processing time of ETL
processes by increasing computational power of an ETL server
and by applying parallel processing of data. This can be achieved
by: (
On top of this, an ETL task re-ordering may be used to produce a more eficient processes. In the simplest case, called push down, the most selective tasks are moved towards the beginning of an ETL process (towards data sources) to reduce a data volume (I/O) as soon as possible [26]. IBM extended this technique into balanced optimization, where some tasks are moved into DSs and some are moved into a DW, to be executed there, cf. Section 3. 2.2
Research approaches to optimizing ETL execution can be
categorized as: (
2.2.1 Cost-based, task re-ordering, and parallelization. In [19, 20] the authors apply the MapReduce framework to process ETL workflows to feed in parallel dimensional schemas (star, snowflake, and SCDs). The authors provide some processing skeletons for MapReduce for dimensions of diferent sizes and structures. A system for optimizing MapReduce workloads was presented in [11]. The optimization uses: sharing of data flows, materialization, and physical data layouts. However, neither an optimization technique nor a cost function were explained.
[18] proposes to parallelize ETL process P in three steps. First, P is partitioned into linear sub-processes LPi . Second, in each LPi , input data are partitioned horizontally into n disjoint partitions (n is parameterized) and each partition is processed in parallel by a separate thread. Third, for tasks with a heavy computational load, an internal multi-threading parallelization may be applied, if needed.
[10] assumes that an ETL process is assigned an estimated execution cost and a more eficient variant of this process is produced by task re-ordering, but the authors do not provide any method for the re-ordering. The main goal of [10] is to identify a set of statistics to collect for this kind of optimization. The constraint is that this set must be: minimal and applicable to estimate costs of all possible task re-orderings. Moreover, the cost of collecting statistics must be minimal. The statistics include: a cardinality of table Ti , attribute histograms of Ti , and a number of distinct values of attributes of Ti . A proposed cost function includes: data statistics, CPU speed, disk-access speed, and memory usage. The authors proposed cost functions for the following ETL tasks (all expressed via SQL): select, project, join, group-by, and transform. To find the set of statistics, which is an NP-hard problem, the authors use linear programming.
The approach presented in [24] goes one step further, as it proposes a specific performance optimization technique by task re-ordering. To this end, each workload gets assigned an execution cost. The main components of the cost formula are time and data volume. The five following workflow transformation operations were proposed: swap - changing the order of two tasks, factorize - combining tasks that execute the same operation on diferent flows, distribute - splitting the execution of a task into n parallel tasks, merge - merging two consecutive tasks, and its reverse operation - split [23]. For each of these operations, criteria for correct workflow transformations were proposed. Finally, a heuristic for pruning the search space of all possible workflow transformations (by task re-ordering) was proposed, with the main goal to filter data as soon as possible. In [15], the re-ordering of operators is based on their semantics, e.g., a highly selective operator would be placed (re-ordered) at the beginning of a workflow.
In the same spirit, workflow transformations were proposed in [14] for the purpose of being able to reuse existing data processing workflows and integrate them into other workflows.
The approach proposed in [16] draws upon the contributions of [24]. In [16], possible tasks re-orderings are constrained by means of a dedicated structure called a dependency graph. This approach optimizes only linear workflows. To this end, a nonlinear workflow is split into linear ones (called groups), by means of pre-defined split rules. Next, parallel groups are optimized independently by task re-ordering - tasks can be moved between adjacent groups, and adjacent groups can be merged. The drawback of this approach is however, that the most selective tasks can be moved towards the end of a workflow as the result of a re-ordering.
2.2.2 ETL with UDFs.
None of the approaches described in Sections 2.2.1 and 2.1
supports the optimization of ETL processes with user-defined
functions. In the techniques based on the re-ordering of operators,
the semantics of the operators must be known. Such a
semantics is known and understood for traditional operators based on
the relational algebra. However, the semantics of user defined
functions is typically unknown. For this reason, UDFs have been
handled by solutions that require: either (
In [12, 13], UDFs are treated as black boxes. The framework
consists of the Nephele execution engine and the PACT compiler
to execute UDFs, based on the PACT programming model [5]
(PACT is a generalization of MapReduce). PACT can use hints to
execute a given task in parallel. Such hints are later exploited by
a cost-based optimizer to generate parallel execution plans. The
optimization is based on: (
A framework proposed in [7], called SQL/MR, enables a parallelization of UDFs in a massively-parallel shared-nothing database. To achieve a parallelism, SQL/MR requires a definition of the Row and Partition functions and corresponding execution models for the SQL/MR function instances. The Row function is described as an equivalent to a Map function in MapReduce. Row functions perform row-level transformation and processing. The execution model of the Row function allows independent processing of each input row by exactly one instance of the SQL/MR function. The Partition function is similar to the Reduce function in MapReduce. Exactly one instance of a SQL/MR function is used to independently process each group of rows defined by the PARTITION BY clause in a query. Independent processing of each partition allows the execution engine to achieve parallelism at the level of a partition. The dynamic cost-based optimization is enabled for re-orderings of UDFs.
An optimizer proposed in [21] rewrites an execution plan
based on: (
The aforementioned approaches require understanding the semantics of an UDF (a black-box) by means of either parsing an UDF code, or applying certain coding style, or using certain keywords, or using parallelization hints. Moreover, the approaches do not provide means of analyzing an optimal architecture conifguration (e.g., the number of nodes in a cluster) or a degree of parallelism.
In [2] the authors developed a framework for using pre-defined generic and specific code skeletons for writing programs to be executed in a parallel environment. The code is then generated automatically and the process is guided by configuration parameters that define the number of nodes a program is executed on.
In [1, 3] the authors proposed an overall architecture for executing UDFs in a parallel environment. The architecture was further extended in [4] with a model for optimizing a cluster conifguration, to provide a sub-optimal performance of a given ETL process with UDFs. The applied model is based on the multiple choice knapsack problem and the lp_solve library is used to solve the problem (its implementation is available on GitHub1).
The work described in this paper applies: (
The balanced optimization [17, 27] is implemented in IBM
InfoSphere DataStage. Its overall goals are to: (
Moving tasks is controlled by DataStage preferences [17] that guide the DataStage engine to move certain pre-defined tasks into either a DS, or a DW, or execute them in the engine. The following tasks are controlled by the preferences: transformation, ifltering, aggregation, join, look-up, merge, project, bulk I/O, and temporary staging tables. Once a given ETL process has been designed, a user specifies the preferences. Then the ETL process gets optimized by DataStage, taking into account the preferences. An executable version of the process is generated and deployed.
So far, the balanced optimization has been proven to be profitable when applied to structured data sources. Since numerous NoSQL and stream DSs become first class citizens, assessing the applicability of this type of optimization to such data sources may have a real business value. The work presented in this paper is the first one to assess the balanced optimization on a stream data source. 4
The goal of the experiments was to assess if the balanced
optimization in IBM DataStage may increase the performance of an
ETL process ingesting data from a stream data source, run by
Kafka. Such a software architecture is frequently used by IBM
customers. In particular, we were interested in figuring out: (
To this end a pico-cluster composed of four physical workstations was built. Each workstation included: a 4-core CPU 3GHz, 16GB RAM, 256GB HDD, and was run under Linux RedHat. Two workstations run Kafka and the other two run the IBM InfoSphere DataStage ETL server. The ETL processes were designed using DataStage Designer. The system was fed with rows from table Store_Sales (1.5GB), from the TPC-D benchmark. The performance statistics were measured on each node by iostat. In each scenario 12 experiments were run. The lowest and the highest measurements were discarded and the average of the remaining 10 was computed and is shown in all the figures. Due to the space constraints, we present only selected experimental scenarios.
Notice that the goal of these experiments was to assess the behaviour of the system only at the border between Kafka and DataStage. For this reason, the ETL processes used in the experiments included elementary tasks/components, like: Kafka connector, column import, filtering, aggregation, and storing output in a file. An example process splitting a data flow is shown in Figure 1. RecordCount is a parameter of a connector from DataStage to Kafka. It defines a number of rows (a rowset) that are read from a topic. After a rowset is read, the ETL connector outputs the rowset to the ETL process for further processing.
The value of RecordCount in the experiments ranged from 1 to
’ALL’. In the latter case, the whole test data set was read before
starting the ETL process (in the experiment 11 million of rows).
The elapsed times of processing the whole data set are shown
in Figure 2. The value of the standard deviation ranges from
0.5 to 2.5. From the chart we observe that: (
Kafka allows to split streaming data into partitions, each of which is read by a consumer. As the number of partitions may strongly influence the performance of the whole ETL process 2, figuring out how performance is afected by the number of partitions is of great interest.
To assess the impact of the number of partitions on processing time, in our experiments the number of partitions ranged from 1 to 8. The number of partitions equaled to the number of Kafka consumers. The performance results are shown in Figure 3. Here we show only the results for RecordCount = ALL. As we can observe, the elapsed processing time of the ETL process varies and 2https://www.confluent.io/blog/how-choose-number-topics-partitions-kafka-cluster the best performance was achieved for the number of partitions equaled to 2 and 3. The standard deviation ranges from 0.5 to 1.3.
A detailed analysis of I/O (not shown here due to space constraints) reveals that the number of I/O increases with the increasing number of partitions. Similar increase was observed also in a CPU time. That may be some of the factors that influence the characteristics shown in Figure 3. 4.3
In this experiment we assessed the elapsed ETL processing time w.r.t. the selectivity of a data ingest from Kafka (the selectivity is defined as: the number of rows ingested / the total number of rows). Two scenarios were implemented. In the first one, operation push down was applied to move data filtering into Kafka (available in library Kafka Streams). In the second one, data were ifltered by a standard pre-defined task in DataStage. The results for RecordCount=10 and RecordCount=ALL are shown in Figure 4. The results labeled with sufix PD denote the scenario with operation push down applied. The standard deviation ranges from 0.1 to 2.8.
The results clearly show that a wrong combination of RecordCount with applied push down can decrease the overall performance of an ETL process, cf. RecordCount=ALL PD (with push down) vs. RecordCount=ALL (without push down), for selectivity=90%. From other experiments (not presented in this paper) we observed that values of RecordCount ≤ 20, caused that push down ofered stable better performance for the whole range of selectivities. 10
50 selectivity [%] 90 4.4
The purpose of a flow split is to divide a data flow into multiple
lfows. In our case, we split the input flow into two, using a
parameterized split ratio: from 10% of data in one flow to an even
split. The flow split was executed in: (
RecordCount=10 RecordCount=ALL PD
RecordCount=ALL 50/50 40/60
30/70 data flow split ratio [%] 20/80 10/90
The characteristics show the performance improvement while applied push down for RecordCount=10, cf. bars labeled RecordCount=10 PD vs. RecordCount=10. From other experiments (not discussed in this paper) we observed that the maximum value of RecordCount that improved the performance is 10. From the analysis of detailed data on CPU and I/O, we tentatively draw a conclusion that the observed behaviour is caused by Kafka that was not able to deliver data on time into DataStage. 4.5
In this experiment, count is used to count records in groups.
The number of groups ranges from 10 to 90. The aggregation is
executed in two variants: (
The analysis of CPU and I/O reveals that push down caused higher CPU usage times and higher I/O, which resulted in much worse performance of the aggregation in Kafka than in DataStage. 180 160 140 60 40 20
In practice, data integration architectures are frequently built using IBM DataStage, NoSQL (HBase, Cassandra), and stream DSs (Kafka). The balanced optimization available in DataStage ofers performance improvements for relational data sources, however, its characteristics on NoSQL and stream DSs are unknown. For this reason, a project was launched at IBM to discover these characteristics, with a goal to build an execution optimizer for a standard and multi-cloud architectures [6], based on a learning recommender.
In this paper we presented the experimental evaluation of the balanced optimization applied to a stream DS run by Kafka, within a joint project of IBM and Poznan University of Technology. To the best of our knowledge, it is the first project on analyzing the possibility of using this type of ETL optimization on nonrelational DSs.
From this evaluation, the most interesting observations of a
real business value are as follows. First, Kafka turned out to be
a bottleneck for push down applied to: (
Even though, the aforementioned observations cannot be generalized (they apply to this particular experimental setting), they turned out to be of practical value. First, they were very well received by the Executive Management @IBM. Second, the observations were integrated into a knowledge base of IBM Software Lab and have already been used for multiple proofs of concept.
The next phase of this project will consist in: (
Acknowledgements. The work of Robert Wrembel is supported
by: (