Calculating Cumulative totals/Running totals in Hive

Hi All,

Many of us might have come across a requirement to implement running totals. Below are a few of the implementations.

Consider a bank maintains account and balance details as below.

Running total should give the sum of balance from beginning to current month.

SELECT acc_no, acc_type,Yr_Mnth, bal, SUM(bal) OVER (PARTITION BY (acc_no,acc_type) ORDER BY Yr_Mnth)  as running_total FROM bnk_bal;

If rolling has to happen to a fixed number of rows, we can use RANGE BETWEEN.

ex: Below query will give the running total for every 3 months.

SELECT acc_no, acc_type,Yr_Mnth, bal, SUM(bal) OVER (PARTITION BY (acc_no,acc_type) ORDER BY Yr_Mnth RANGE BETWEEN PRECEDING 2 ROWS AND CURRENT ROW)  as running_total FROM bnk_bal;

Flume Tutorial

Download and import Hortonworks sandbox into VM.

Start the VM once imported and open the URL shown on screen once the sandbox is ready.

Hortonworks Sandbox URL

Enter the above URL in browser and open Ambari to make sure Flume service is running

Flume Service check

Connect to sandbox from any ssh client like putty. (I've used Mobaxterm here)

HWX sandbox connection from ssh client

Create Config file as below in any directory of your choice

Flume Configuration

Change the directory to /usr/hdp/current/flume-server/bin

Execute the below command to start Flume agent i.e agt1 in above conf file. and connect to related channel & sink.

flume-ng agent --conf conf --conf-file /root/flume_conf.conf --name agt1 -Dflume.root.logger=INFO,console

Flume stats

Make sure that Sink, Source are started and connected with an active channel. (observe the above diagram)

Now, go to the source/spool directory location given in conf directory and create a sample file.
you will observe the file getting consumed immediately after the creation.

TestFile creation in spool directory

Started moving the file

Process completed

Source file renamed

File created in HDFS (sink directory)

Observation:  the file in source directory will be renamed to <filename>.COMPLETED

Starting and stopping NiFi jobs from Unix

Hello All,

Below is some info regarding the execution of NiFi flow/process group from Unix using Curl.

Find the processor ID and the group id i.e UUID from the NiFi User interface.
Use below URL in browser to find the process group ID and the related parent ID. Also, collect the version#

http://<hostname>:8080/nifi-api/controller/process-groups/root?verbose=true

hostname - generally it is localhost or any other specific URL provided by your organization.

only version number can be obtained with below command.
curl --request GET http://<hostname>:8080/nifi-api/controller/revision/

To Start a Process group:

curl -i -X PUT -H 'Content-Type: application/json' -d '{"revision":{"version":"<version#>"},"processGroup":{"id":"<ProcessGroupID>","parentGroupId":"<ParentID>","running":"true"}}' http://<hostname>:8080/nifi-api/controller/process-groups/<ProcessGroupID>/

To Stop a Process group, change the value for running to false

Handling skew data in Hive

Method 1: 
Identify the skew value and run the job for that value separately.
ex: If cust_id=100 has skew problem, then divide the records into cust_id=100 and cust_id!=100. then run the individual jobs.
Method 2:
Identify the column creating skew. If it is used for join, try to reduce the skew using multiple columns and use it in join.
Method 3:
Modify the join key(Salting). A simple approach that i follow some times is appending (key%3) at the end of key. ex: key1 can be divided into multiple values like key1_1,key1_2 etc,. This will help distribute the keys.
Method 4:
Divide the data into chunks and execute.

Sqooping from/to MYSQL and HDFS

Sqoop Import

From MYSQL table to HDFS

sqoop import --connect jdbc:mysql://127.0.0.1/export --username root --driver com.mysql.jdbc.Driver --table exported --target-dir '/user/horton/exported' -m 1

From MYSQL table to Hive

sqoop import --hive-import --connect jdbc:mysql://127.0.0.1/export --username root --driver com.mysql.jdbc.Driver --table exported --hive-table test.exported3 -m 1

From MYSQL table to Hive schema defined.

sqoop import --hive-import --connect jdbc:mysql://127.0.0.1/export --username root --driver com.mysql.jdbc.Driver --table exported --hive-table test.exported3 -m 1 --map-column-hive 'row_key=int,value=string,ts=timestamp';

From MYSQL table to Hive with delimiters

sqoop import --hive-import --connect jdbc:mysql://127.0.0.1/export --username root --driver com.mysql.jdbc.Driver --table exported --hive-table test.exported2 -m 1 --fields-terminated-by ','

Free form query import

sqoop import --connect jdbc:mysql://127.0.0.1/export --username root --driver com.mysql.jdbc.Driver --query 'select * from exported where $CONDITIONS' --target-dir '/user/horton/test_query' -m 1

Incremental import into HDFS

sqoop import --connect jdbc:mysql://127.0.0.1/export --username root --driver com.mysql.jdbc.Driver --table exported --target-dir '/user/horton/test_incremental' -m 1 --check-column ts --incremental lastmodified --last-value '2017-01-24 23:11:16.0' -merge-key row_key

Incremental import into Hive

sqoop import --hive-import --connect jdbc:mysql://127.0.0.1/export --username hive --password hive --driver com.mysql.jdbc.Driver --table exported --hive-table test.exported1 -m 1 --check-column ts --incremental lastmodified --last-value '2017-01-24 23:11:10.0' -merge-key row_key

Incremental import into HDFS from Query

sqoop import --connect jdbc:mysql://127.0.0.1/export --username root --driver com.mysql.jdbc.Driver --query 'select * from exported where $CONDITIONS' --target-dir '/user/horton/test_incremental/t2' -m 1 --check-column ts --incremental lastmodified --last-value '2017-01-24 23:11:16.0' -merge-key row_key

Incremental import into Hive from Query is not supported.

Sqoop Export

Insert/export from HDFS

sqoop export --connect jdbc:mysql://127.0.0.1/export --username root --driver com.mysql.jdbc.Driver --table exported --export-dir /apps/hive/warehouse/test.db/exported1  -m 1 --input-fields-terminated-by '\001'

Insert/export from Hive

sqoop export --connect jdbc:mysql://127.0.0.1/export --username root --driver com.mysql.jdbc.Driver --table exported --hcatalog-database test --hcatalog-table exported3 --input-fields-terminated-by '\001' -m 1

Update from HDFS

sqoop export --connect jdbc:mysql://127.0.0.1/export --username root --driver com.mysql.jdbc.Driver --table exported --export-dir /apps/hive/warehouse/test.db/exported1  -m 1 --update-key row_key --input-fields-terminated-by '\001'

Using Hcatalog

sqoop export --connect jdbc:mysql://127.0.0.1/export --username root --driver com.mysql.jdbc.Driver --table exported --hcatalog-database test --hcatalog-table exported -m 1

Update from Hive using Hcatalog is not supported yet. Getting Java exception and there are open tickets on it.

MapReduce V2 Job Flow

Mapreduce job flow on YARN involves below components.

• A Client node, which submits the Mapreduce job.
• The YARN Resource Manager, which allocates the cluster resources to jobs.
• The YARN Node Managers, which launch and monitor the tasks of jobs.
• The MapReduce Application Master, which coordinates the tasks running in the MapReduce job. The
• application master and the MapReduce tasks run in containers that are scheduled by the resource
• manager, and managed by the node managers.
• The HDFS file system is used for sharing job files between the above entities.

Job Start up:

• The call to Job.waitForCompletion() in the main driver class is where all the execution starts. The driver is the only piece of code that runs on our local machine, and this call starts the communication with the Resource Manager
• Retrieves the new Job ID or Application ID from Resource Manager.
• The Client Node copies Job Resources specified via the -files, -archives, and -libjars command-line arguments, as well as the job JAR file on to HDFS.
• Finally, Job is submitted by calling submitApplication() method on Resource Manager.
• Resource Manager triggers its sub-component Scheduler, which allocates containers for mapreduce job execution. Then Resource Manager starts Application Master in the container provided by the scheduler. This container will be managed by Node Manager from here on wards.

Input Split:

• In this phase, HDFS splits the input files into equal sized chunks or segments based on minimum split size (mapreduce.input.fileinputformat.split.minsize) property .
• Each file segment or split is passed to a unique map task if file is splittable. If File is not splittable then entire file will be provided as input to a single map task.
• These map tasks are created by Mapreduce Application Master (MRAppMaster Java Class) and reduce tasks are also created by application master based on mapreduce.job.reduces property.

Role of an Application Master:

• Before starting any task, Job setup method is called to create job’s output directory for job’s OutputCommitter.
• As noted above, Both map tasks and reduce tasks are created by Application Master.
• If the submitted job is small, then Application Master runs the job in the same JVM on which Application Master is running. It reduces the overhead of creating new container and running tasks in parallel. These small jobs are called as Uber tasks.
• Uber tasks are decided by three configuration parameters, number of mappers <= 10, number of reducers <= 1 and Input file size is less than or equal to an HDFS block size. These parameters can be configured via mapreduce.job.ubertask.maxmaps , mapreduce.job.ubertask.maxreduces , and mapreduce.job.ubertask.maxbytes properties in mapred-site.xml.
• If job doesn’t qualify as Uber task, Application Master requests containers for all map tasks and reduce tasks.

Task Execution:

• Once Containers assigned to tasks, Application Master starts containers by notifying its Node Manager.
• Node Manager copies Job resources (like job JAR file) from HDFS distributed cache and runs map or reduce tasks.
• Running Tasks, keep reporting about the progress and status (Including counters) of current task to Application Master and Application Master collects this progress information from all tasks and aggregate values are propagated to Client Node or user.

To date calculations in SQL

CREATE TABLE m2dt(sale_amt INT, sale_dt DATE);
INSERT INTO m2dt VALUES(30,'2015-01-01'),(50,'2015-02-01'),(100,'2016-01-01'),(50,'2016-01-10'),(30,'2016-01-03');

Month to Date calculation:
SELECT 
SALE_DT,
SUM(sale_amt) OVER (PARTITION BY TRUNC(sale_dt,'MM') ORDER BY STRING(sale_dt)) AS amt 
FROM m2dt;

Year to Date calculation:
SELECT 
SALE_DT,
SUM(sale_amt) OVER (PARTITION BY YEAR(sale_dt) ORDER BY TRUNC(sale_dt,'MM')) AS amt 

FROM m2dt;