Reading XML files using Hive

Scenario 1:
Consider an XML file as below

<row Id="1" Reputation="100" creationDate="2008-07-31T14:22:31.317" DisplayName="Test 1" LastAccessDate="2016-12-10T22:12:46.367" WebsiteUrl="http://www.joelonsoftware.com/" Location="New York, NY" />
<row Id="2" Reputation="250" creationDate="2008-07-31T14:22:31.317" DisplayName="Test User2" LastAccessDate="2016-12-10T22:12:46.367" WebsiteUrl="http://www.test.com/" Location="Phoenix, AZ"  Age="25"/>

Download Jar:

wget http://search.maven.org/remotecontent?filepath=com/ibm/spss/hive/serde2/xml/hivexmlserde/1.0.5.3/hivexmlserde-1.0.5.3.jar

From Hive CLI/Beeline:

add jar <above_jar_location>;

CREATE EXTERNAL TABLE test_xml(
Id int,
Reputation int,
creationDate timestamp,
displayName string,
location string,
age int
)
ROW FORMAT SERDE 'com.ibm.spss.hive.serde2.xml.XmlSerDe'
WITH SERDEPROPERTIES (
"column.xpath.id"="/row/@Id",
"column.xpath.Reputation"="/row/@Reputation",
"column.xpath.creationDate"="/row/@CreationDate",
"column.xpath.displayName"="/row/@DisplayName",
"column.xpath.location"="/row/@Location",
"column.xpath.age"="/row/@Age"
)
STORED AS
INPUTFORMAT 'com.ibm.spss.hive.serde2.xml.XmlInputFormat'
OUTPUTFORMAT 'org.apache.hadoop.hive.ql.io.IgnoreKeyTextOutputFormat'
LOCATION '/path/to/xml/file'
TBLPROPERTIES (
"xmlinput.start"="<row ",
"xmlinput.end"="/>"
);

Scenario 2:

Consider Below XML:

<CATALOG>

<BOOK>

<TITLE>Hadoop Defnitive Guide</TITLE>

<AUTHOR>Tom White</AUTHOR>

<CURRENCY>USD</CURRENCY>

<PRICE>34</PRICE>

<YEAR>2017</YEAR>

</BOOK>

<BOOK>

<TITLE>Programming with Spark</TITLE>

<AUTHOR>I don't know</AUTHOR>

<CURRENCY>USA</CURRENCY>

<PRICE>29</PRICE>

<YEAR>2018</YEAR>

</BOOK>

</CATALOG>

Create table as below:

CREATE EXTERNAL TABLE books (title string, price float,currency string)

ROW FORMAT SERDE 'com.ibm.spss.hive.serde2.xml.XmlSerDe'

WITH SERDEPROPERTIES (

"column.xpath.title"="/BOOK/TITLE/text()",

"column.xpath.currency"="/BOOK/CURRENCY/text()",

"column.xpath.price"="/BOOK/PRICE/text()")

STORED AS INPUTFORMAT 'com.ibm.spss.hive.serde2.xml.XmlInputFormat'

OUTPUTFORMAT 'org.apache.hadoop.hive.ql.io.IgnoreKeyTextOutputFormat'

LOCATION '/path/to/xml'

TBLPROPERTIES ("xmlinput.start"="<BOOK","xmlinput.end"= "</BOOK>");

**** Do not specify the Root node i.e <CATALOG> in TBLPROPERTIES****

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

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.

SCD Type1 implementation in Hive

Consider below tables with partitions.

Stage table Final Table
p1 p1
p2 p2
p5 p3
p4

stage_partition left join final_partition on partition columns
Set1: Consider the data from stage table where there is a match - Updates are available for the data in these partitions
Set2: Consider the data from stage table where there is no match - New records (Inserts)

create table final_partition_updates as select * from final_table where partition_col in (set1);
create table stage_partition_updates as select * from stage_table where partition_col in (set1);

WITH pkeyJoin as select a.*,b.* from final_partition_updates a full join stage_partition_updates b
on a.pkey = b.pkey

create table upserts as
select b.* from pkeyJoin where a.pkey is not null  --Updates
UNION ALL
select b.* from pkeyJoin where a.pkey is null; --Inserts

Insert into table upserts select * from stage_table where partition_col in (set2);

Insert overwrite final_table PARTITION(partition column) SELECT * from upserts;