I wrote, separately, a Hadoop job to calculate running averages of about
2000 stock tickers over a 180 day period as well as a Hive script which
performs equivalent functionality. I have been using Amazon Elastic
MapReduce as my platform for running these jobs. I have been trying for a
while to get my Hive script to perform well when spread over many nodes,
but cannot seem to get the Hive script to perform nearly as well as the
Hadoop job. (The Hadoop job takes about an hour to run through all of my
tickers, whereas the Hive job takes over an hour just to run about 1/8 of
them and I cannot even seem to get it to finish when I run it for a larger
number of tickers.) I also have not seen large gains when running my Hive
job using a larger number of hosts. I've been trying to tinker with
settings, examine the query plans of my queries, attempt many modifications
of my queries, but have not seen great gains in performance.
Here is my code. Can you help me identify potential problem points and
ways I can improve these queries, especially so they distribute well when
run on multiple hosts. I tried to add comments where appropriate to make
it clear what I was doing in each step. Please note there are about 2000 *
180 = 360,000 rows in the raw symbol table.
Please help, I am quite stuck on this! Feel free to ask any questions for
which you would like clarification.
Here is my script:
ADD JAR ${INPUT}/market-data.jar ;
ADD JAR ${INPUT}/HiveJars/derby.jar;
set hive.stats.autogather=false;
set hive.exec.dynamic.partition.mode=nonstrict;
set hive.exec.dynamic.partition=true;
set hive.exec.reducers.bytes.per.reducer=1000000000;
set hive.exec.max.dynamic.partitions.pernode=200000;
set hive.exec.max.dynamic.partitions=200000;
set hive.exec.max.created.files=1000000;
-- Note ${INPUT} is the S3 URL to where my scripts and input files are
stored. ${INPUT}/hiveinput/output contains separate folders labeled
symbol=[ticker symbol] so that
-- they can be imported into a partitioned table. The files in these
folders contain the ticker prices of each of the stocks over a 180 day
period obtained from Yahoo Finance
CREATE EXTERNAL TABLE raw_symbols
(dt STRING, open STRING, high STRING, low STRING, close STRING,
volume STRING, adj_close STRING)
PARTITIONED BY (symbol STRING)
ROW FORMAT DELIMITED FIELDS TERMINATED BY ',' LINES TERMINATED BY '\n'
STORED AS TEXTFILE
LOCATION '${INPUT}/hiveinput/output' ;
-- Elastic MapReduce requires you to execute this command to create all of
the dynamic partitions corresponding to the stock tickers
ALTER TABLE raw_symbols RECOVER PARTITIONS;
-- This is simply loading a table with the sequence 1 through 90. I
actually couldn't find anything in Hive to create a simple integer
sequence. So this table is loaded with
-- this sequence
CREATE EXTERNAL TABLE day_seq_orig
(day INT)
ROW FORMAT DELIMITED LINES TERMINATED BY '\n'
STORED AS TEXTFILE
LOCATION '${SEQFILE}';
-- A temporary table to contain the distinct list of dates for which we
have stock prices, should be 180 dates for the 180 days for which we are
getting info
CREATE TABLE distinct_dt
(dt STRING)
STORED AS SEQUENCEFILE;
-- ${SAMPLE_SYMBOL} is just one of my symbols. Since the same sequence of
dates applies to all tickers, this gives me an easy and quick way to get
the range of dates
INSERT OVERWRITE TABLE distinct_dt
SELECT distinct dt
FROM raw_symbols
WHERE raw_symbols.symbol = '${SAMPLE_SYMBOL}';
CREATE TABLE day_seq
(current_date STRING, original_date STRING)
STORED AS SEQUENCEFILE;
-- We are calculating a 90 day rolling average for each stock ticker price,
so I want to get those dates within 90 days of each of the dates for which
I am getting stock ticker prices
INSERT OVERWRITE TABLE day_seq. date_add is a custom Hive function I
implemented; it does just what it implies: it adds an integral number to a
date.
SELECT date_add(dt, day), dt
FROM distinct_dt
JOIN day_seq_orig ON (1=1);
CREATE TABLE converted_symbols
(
symbol STRING,
original_date STRING,
close FLOAT
)
STORED AS SEQUENCEFILE;
CREATE TABLE converted_symbols_cf
(
original_date STRING,
close FLOAT
)
PARTITIONED BY (symbol STRING)
STORED AS RCFILE;
CREATE TABLE converted_symbols_cf_abbr
(
original_date STRING,
close FLOAT
)
PARTITIONED BY (symbol STRING)
STORED AS RCFILE;
CREATE TABLE dt_partitioned_symbols
(
symbol STRING,
close FLOAT
)
PARTITIONED BY (original_date STRING)
STORED AS SEQUENCEFILE;
CREATE TABLE joined_symbols
(
original_date STRING,
current_date STRING, close FLOAT
)
PARTITIONED BY (symbol STRING)
STORED AS SEQUENCEFILE;
-- Take the raw symbol stock prices, which did no data conversions, and
convert the price to a float and put into a temp table
INSERT OVERWRITE TABLE converted_symbols_cf
PARTITION (symbol)
SELECT
dt, cast(close AS FLOAT), symbol
FROM raw_symbols
DISTRIBUTE BY symbol;
-- Remove some erroneous header rows and put in another temp table
INSERT OVERWRITE TABLE converted_symbols_cf_abbr
PARTITION (symbol)
SELECT
original_date, close, symbol
FROM converted_symbols_cf
WHERE original_date != 'Date'
DISTRIBUTE BY symbol;
set hive.exec.max.dynamic.partitions.pernode=200000;
set hive.exec.max.dynamic.partitions=200000;
-- Join my date table with my latest stock ticker price table. Use a map
join because the day_seq table only has about 18,000 rows
INSERT OVERWRITE TABLE joined_symbols
PARTITION (symbol)
SELECT /*+ MAPJOIN(day_seq) */ day_seq.original_date, day_seq.current_date,
close, symbol
FROM converted_symbols_cf_abbr
JOIN day_seq ON (converted_symbols_cf_abbr.original_date =
day_seq.original_date)
distribute by symbol;
CREATE EXTERNAL TABLE symbols_with_avgs
(current_date STRING, close FLOAT)
PARTITIONED BY(symbol STRING)
ROW FORMAT DELIMITED FIELDS TERMINATED BY ',' LINES TERMINATED BY '\n'
STORED AS TEXTFILE
LOCATION '${OUTPUT}/hiveoutput' ;
-- Group and calculate my averages and output
INSERT OVERWRITE TABLE symbols_with_avgs
PARTITION (symbol)
SELECT current_date, avg(close), symbol
FROM joined_symbols
GROUP BY symbol, current_date
DISTRIBUTE BY symbol;
