Tuesday Oct 22, 2013

How to Load Oracle Tables From Hadoop Tutorial (Part 5 - Leveraging Parallelism in OSCH)

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 words, it 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

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

TheCount

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.


Wednesday Oct 16, 2013

Loading from Multiple Data Sources with Oracle Loader for Hadoop

Oracle Loader for Hadoop can be used to load data from multiple data sources (for example Hive, HBase), and data in multiple formats (for example Apache weblogs, JSON files).   There are two ways to do this:

(1) Use an input format implementation.  Oracle Loader for Hadoop includes several input format implementations.  In addition, a user can develop their own input format implementation for proprietary data sources and formats.

(2) Leverage the capabilities of Hive, and use Oracle Loader for Hadoop to load from Hive.

These approaches are discussed in our Oracle Open World 2013 presentation

Thursday Aug 15, 2013

Parallelism while Querying and Loading Data using Oracle SQL Connector for HDFS

For fast performance Oracle SQL Connector for HDFS (OSCH) can use parallel threads in the database to read data files on HDFS.   This post discusses the settings in OSCH that are relevant for using parallel threads, and some heuristics to set them.

First, a quick background on ‘location files.’ OSCH generates an external table that is used to query or load data from HDFS. This external table has a LOCATION clause with a list of location files (see example of a generated external table at the end of this post). OSCH populates the location files with URIs of the data files on HDFS. When there are multiple location files and multiple data files, OSCH uses load balancing to distribute the files (by size) among the location files. Data files are not split up, but the URIs are distributed by data file size as evenly as possible. See the documentation on location files in the Oracle® Big Data Connectors User's Guide Release 2 (2.2) for more details.

Number of Location Files

Degree of Parallelism (DOP) is the number of parallel threads used by the database to read the data. When reading data from external tables, the DOP specified in the database has to be supported by the relevant number of location files in the external table. If there is only one location file, there will be no parallelism while reading the data. The default value for the number of location files is 4. Higher numbers to support a higher DOP can be specified using the locationFileCount property. Keep in mind the resources available in the database while determining the number of parallel threads to use.

Specifying locationFileCount while Using OSCH

As a -D parameter: -D oracle.hadoop.exttab.locationFileCount=256

In a configuration file:

<property>
<name>oracle.hadoop.exttab.locationFileCount</name>
<value>256</value>
</property>

A value of 256 will result in an external table with 256 location files, which in turn means that 256 parallel threads can be used to access the data.

Number of Data Files

If the number of data files accessed is less than the number of location files, the number of parallel threads will be constrained by the number of populated location files. If possible, split up the data files to support higher numbers of parallel threads. If there are more data files than location files OSCH will load balance to evenly distribute data files by size across location files. A good heuristic: the number of data files should be a multiple of the number of location files.

Reading from a Hive table

If OSCH is accessing data from a Hive table with a single underlying data file, a bucketed Hive table can be created to split the input files. If there are more buckets than location files OSCH will load balance. A good heuristic: the number of buckets should be a multiple of the number of location files.

Example on the Oracle Big Data Appliance

Data file size: 1 TB.

If the data is in a Hive table, a bucketed Hive table can be created with 2048 buckets (each file will have 2 blocks on the BDA which has a block size of 256MB). A locationFileCount value of 256 will result in each thread reading about 8 files.

Note on enabling parallel query in the database

Parallel query has to be enabled on the database for data files to be accessed in parallel from HDFS. From the Oracle® Big Data Connectors User's Guide Release 2 (2.2):

Enable parallel query with this SQL command:

ALTER SESSION ENABLE PARALLEL QUERY;

Before loading the data into an Oracle database from the external files created by Oracle SQL Connector for HDFS, enable parallel DDL:

ALTER SESSION ENABLE PARALLEL DDL;

Before inserting data into an existing database table, enable parallel DML with this SQL command:

ALTER SESSION ENABLE PARALLEL DML;

Hints such as APPEND and PQ_DISTRIBUTE also improve performance while inserting data.

Example external table generated by OSCH (LOCATION clause and location files in red)

CREATE TABLE "HDFSUSER"."TEXT_ORA_EXT_TAB"

( "CUST_ID" VARCHAR2(4000),

"MOVIE_ID" VARCHAR2(4000),

"GENRE_ID" VARCHAR2(4000),

"TIME_ID" VARCHAR2(4000),

"RECOMMENDED" VARCHAR2(4000),

"ACTIVITY_ID" VARCHAR2(4000),

"RATING" VARCHAR2(4000),

"SALES" VARCHAR2(4000)

)

ORGANIZATION EXTERNAL

( TYPE ORACLE_LOADER

DEFAULT DIRECTORY "ORACLETEST_DIR"

ACCESS PARAMETERS

( RECORDS DELIMITED BY 0X'0A'

CHARACTERSET AL32UTF8

STRING SIZES ARE IN CHARACTERS

PREPROCESSOR "OSCH_BIN_PATH":'hdfs_stream'

FIELDS TERMINATED BY 0X'09'

MISSING FIELD VALUES ARE NULL

(

"CUST_ID" CHAR(4000),

"MOVIE_ID" CHAR(4000),

"GENRE_ID" CHAR(4000),

"TIME_ID" CHAR(4000),

"RECOMMENDED" CHAR(4000),

"ACTIVITY_ID" CHAR(4000),

"RATING" CHAR(4000),

"SALES" CHAR(4000)

)

)

LOCATION

( 'osch-20130725105856-1332-1’,

  'osch-20130725105856-1332-2',

  'osch-20130725105856-1332-3',

  'osch-20130725105856-1332-4'

 ........

     ........ 

)

)

REJECT LIMIT UNLIMITED

PARALLEL

About

Oracle Loader for Hadoop and Oracle SQL Connector for HDFS

Search

Categories
Archives
« February 2016
SunMonTueWedThuFriSat
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
     
       
Today