How to Load Oracle Tables From Hadoop Tutorial (Part 5 - Leveraging Parallelism in OSCH)
- by Bob Hanckel
Normal
0
false
false
false
EN-US
X-NONE
X-NONE
MicrosoftInternetExplorer4
Using OSCH: Beyond Hello World
In
the previous post we discussed a “Hello World” example for OSCH focusing on the
mechanics of getting a toy end-to-end example working. In this post we are going to talk about how
to make it work for big data loads. We
will explain how to optimize an OSCH external table for load, paying particular
attention to Oracle’s DOP (degree of parallelism), the number of external table
location files we use, and the number of HDFS files that make up the payload. We will provide some rules that serve as
best practices when using OSCH.
The
assumption is that you have read the previous post and have some end to end
OSCH external tables working and now you want to ramp up the size of the loads.
Using OSCH External Tables for Access and Loading
OSCH external tables are no different from any other Oracle external tables. They can be used to access HDFS content using Oracle SQL:
SELECT * FROM my_hdfs_external_table;
or use the same SQL access to load a table in Oracle.
INSERT INTO my_oracle_table SELECT * FROM my_hdfs_external_table;
To speed up the load time, you will want to control the degree of parallelism (i.e. DOP) and add two SQL hints. ALTER SESSION FORCE PARALLEL DML PARALLEL 8; ALTER SESSION FORCE PARALLEL QUERY PARALLEL 8; INSERT /*+ append pq_distribute(my_oracle_table, none) */ INTO my_oracle_table SELECT * FROM my_hdfs_external_table;
There are various ways of either hinting at what level of DOP you want to use. The ALTER SESSION statements above force the issue assuming you (the user of the session) are allowed to assert the DOP (more on that in the next section). Alternatively you could embed additional parallel hints directly into the INSERT and SELECT clause respectively.
/*+ parallel(my_oracle_table,8) *//*+ parallel(my_hdfs_external_table,8) */
Note that the "append" hint lets you load a target table by reserving space above a given "high watermark" in storage and uses Direct Path load. In other doesn't try to fill blocks that are already allocated and partially filled. It uses unallocated blocks. It is an optimized way of loading a table without incurring the typical resource overhead associated with run-of-the-mill inserts. The "pq_distribute" hint in this context unifies the INSERT and SELECT operators to make data flow during a load more efficient.
Finally your target Oracle table should be defined with "NOLOGGING" and "PARALLEL" attributes. The combination of the "NOLOGGING" and use of the "append" hint disables REDO logging, and its overhead. The "PARALLEL" clause tells Oracle to try to use parallel execution when operating on the target table.
Determine Your DOP
It might feel natural to build your datasets in Hadoop, then
afterwards figure out how to tune the OSCH external table definition, but you should start backwards. You should focus on Oracle database, specifically the DOP you want to use
when loading (or accessing) HDFS content using external tables.
The DOP in Oracle controls how many PQ slaves are launched
in parallel when executing an external table. Typically the DOP is something you want to Oracle to control transparently, but for loading content from Hadoop with OSCH, it's something that you will want to control.
Oracle computes the maximum DOP that can be used by an
Oracle user. The maximum value that can be assigned is an
integer value typically equal to the number of CPUs on your Oracle instances,
times the number of cores per CPU, times the number of Oracle instances. For
example, suppose you have a RAC environment with 2 Oracle instances. And suppose that each system has 2 CPUs with
32 cores. The maximum DOP would be 128 (i.e.
2*2*32).
In point of fact if you are running on a production system,
the maximum DOP you are allowed to use will be restricted by the Oracle DBA. This is because using a system maximum DOP
can subsume all system resources on Oracle and starve anything else that is
executing. Obviously on a production
system where resources need to be shared 24x7, this can’t be allowed to happen.
The use cases for being able to run OSCH with a maximum DOP
are when you have exclusive access to all the resources on an Oracle
system. This can be in situations when
your are first seeding tables in a new Oracle database, or there is a time
where normal activity in the production database can be safely taken off-line for
a few hours to free up resources for a big incremental load. Using OSCH on high end machines
(specifically Oracle Exadata and Oracle BDA cabled with Infiniband), this mode
of operation can load up to 15TB per hour.
The bottom line is that you should first figure out what DOP
you will be allowed to run with by talking to the DBAs who manage the
production system. You then use that
number to derive the number of location files, and (optionally) the number of HDFS
data files that you want to generate, assuming that is flexible.
Rule 1: Find out the maximum DOP you will be allowed
to use with OSCH on the target Oracle system
Determining the Number of Location Files
Let’s assume that the DBA told you that your maximum DOP was 8. You want the number of location files
in your external table to be big enough to utilize all 8 PQ slaves, and you
want them to represent equally balanced workloads. Remember location files in OSCH are
metadata lists of HDFS files and are created using OSCH’s External Table tool. They
also represent the workload size given to an individual Oracle PQ slave (i.e. a
PQ slave is given one location file to process at a time, and only it will
process the contents of the location file.)
Rule 2: The size of the workload of a single location
file (and the PQ slave that processes it) is the sum of the content size of the
HDFS files it lists
For example, if a location file lists 5 HDFS files which are
each 100GB in size, the workload size for that location file is 500GB.
The number of location files that you generate is something
you control by providing a number as input to OSCH’s External Table tool.
Rule 3: The number of
location files chosen should be a small multiple of the DOP
Each location file represents one workload for one PQ
slave. So the goal is to keep all slaves
busy and try to give them equivalent workloads. Obviously if you run with a DOP of 8 but
have 5 location files, only five PQ slaves will have something to do and the
other three will have nothing to do and will quietly exit. If you run with 9 location files, then the
PQ slaves will pick up the first 8 location files, and assuming they have
equal work loads, will finish up about the same time. But the first PQ slave to finish its job will then
be rescheduled to process the ninth location file, potentially doubling the
end to end processing time. So for this
DOP using 8, 16, or 32 location files would be a good idea.
Determining the Number of HDFS Files
Let’s start with the next rule and then explain it:
Rule 4: The number of HDFS files should try to be a
multiple of the number of location files and try to be relatively the same
size
In our running example, the DOP is 8. This means that the number of location files
should be a small multiple of 8. Remember that each location file represents a
list of unique HDFS files to load, and that the sum of the files listed in each
location file is a workload for one Oracle PQ slave. The OSCH External Table tool will look in an
HDFS directory for a set of HDFS files to load. It will generate N number of location files
(where N is the value you gave to the tool). It will then try to divvy up the HDFS files and do its best to make sure the
workload across location files is as balanced as possible. (The tool uses a greedy algorithm that grabs
the biggest HDFS file and delegates it to a particular location file. It then looks for the next biggest file and
puts in some other location file, and so on). The tools ability to balance is reduced if HDFS file sizes are grossly out of
balance or are too few.
For example suppose my DOP is 8 and the number of location
files is 8. Suppose I have only 8
HDFS files, where one file is 900GB and the others are 100GB. When the tool tries to balance the load it
will be forced to put the singleton 900GB into one location file, and put each
of the 100GB files in the 7 remaining location files. The load balance skew is 9 to 1. One PQ slave will be working overtime, while
the slacker PQ slaves are off enjoying happy hour.
If however the total payload (1600 GB) were broken up into
smaller HDFS files, the OSCH
External Table tool would have an easier time generating a list where each
workload for each location file is relatively the same. Applying Rule 4 above to our DOP of 8, we could divide the workload into160 files that were approximately 10 GB in size. For this scenario the OSCH External Table tool would populate
each location file with 20 HDFS file references, and all location files would
have similar workloads (approximately 200GB per location file.)
As a rule, when the OSCH External Table tool has to deal
with more and smaller files it will be able to create more balanced loads. How
small should HDFS files get? Not so
small that the HDFS open and close file overhead starts having a
substantial impact. For our performance
test system (Exadata/BDA with Infiniband), I compared three OSCH loads of 1
TiB. One load had 128 HDFS files living
in 64 location files where each HDFS file was about 8GB. I then did the same load with 12800 files where
each HDFS file was about 80MB size. The
end to end load time was virtually the same. However when I got ridiculously small (i.e. 128000 files at about 8MB
per file), it started to make an impact and slow down the load time.
What happens if you break rules 3 or 4 above? Nothing draconian, everything will still
function. You just won’t be taking full advantage
of the generous DOP that was allocated to you by your friendly DBA.
The key point of the rules articulated above is this: if you know that HDFS content is ultimately going to be loaded into Oracle using OSCH, it makes sense
to chop them up into the right number of files roughly the same size, derived
from the DOP that you expect to use for loading.
Next Steps
So far we have talked about OLH and OSCH as alternative
models for loading. That’s not quite
the whole story. They can be used
together in a way that provides for more efficient OSCH loads and allows one to
be more flexible about scheduling on a Hadoop cluster and an Oracle Database to
perform load operations. The next
lesson will talk about Oracle Data Pump files generated by OLH, and loaded using
OSCH. It will also outline the pros and cons of using various load methods. This will be followed up with a
final tutorial lesson focusing on how to optimize OLH and OSCH for use on Oracle's engineered systems: specifically Exadata and
the BDA.
/* Style Definitions */
table.MsoNormalTable
{mso-style-name:"Table Normal";
mso-tstyle-rowband-size:0;
mso-tstyle-colband-size:0;
mso-style-noshow:yes;
mso-style-priority:99;
mso-style-qformat:yes;
mso-style-parent:"";
mso-padding-alt:0in 5.4pt 0in 5.4pt;
mso-para-margin-top:0in;
mso-para-margin-right:0in;
mso-para-margin-bottom:10.0pt;
mso-para-margin-left:0in;
line-height:115%;
mso-pagination:widow-orphan;
font-size:11.0pt;
font-family:"Calibri","sans-serif";
mso-ascii-font-family:Calibri;
mso-ascii-theme-font:minor-latin;
mso-fareast-font-family:"Times New Roman";
mso-fareast-theme-font:minor-fareast;
mso-hansi-font-family:Calibri;
mso-hansi-theme-font:minor-latin;}
