Sqoop with HCatalog and Oozie
Sqoop may use HCatalog to import and export data directly into/from Hive tables. It uses HCatalog to read table’s structure, data formats, partitions and then imports/exports data appropriately. It’s very useful combination for efficient data move, but requires matching column names on both sides. Here’s how to make Sqoop with HCatalog work through Oozie.
Contents
HOW TO
I used CDH 5.8.4 with Kerberos.
Import
In order to import data from relational database into Hive metastore with Sqoop using HCatalog first you need to create a table, i.e. with Hive. Table can be internal or external, partitioned or not, but the column names need to match those on the database side:
|
1 2 3 4 5 6 7 8 9 |
CREATE EXTERNAL TABLE ${schema}.{table_name}( col1 string, col2 bigint, col3 int, col4 string ) partitioned by (partition_column string) stored as sequencefile location '${location_dir}'; |
Note: By design columns in database table must match the ones in created table! It’s because HCatalog does the column matching by names. If some columns won’t match you’ll get error message:
|
1 |
java.lang.NullPointerException at org.apache.hive.hcatalog.data.schema.HCatSchema.get(HCatSchema.java:105) |
Note: Original database column names are converted to lowercase and ‘-‘ characters are replaced with “_” by Sqoop. This has to be taken into consideration while creating Hive table.
Note: Currently Parquet format is not supported for Sqoop with HCatalog – there’s an issue pending.
Alternatively to creating a Hive table, you may choose to delegate Sqoop to create it for you with ‐‐create-hcatalog-table option.
If you want to do the import with Oozie use the Sqoop import action with:
- HCatalog credentials
- ‐‐hcatalog-database and ‐‐hcatalog-table options
- jdbc jar file
- hive-site.xml file
Your hive-site.xml should contain at least following properties:
|
1 2 3 4 5 6 7 8 9 10 11 12 |
<property> <name>hive.metastore.uri</name> <value>${hive_metastore_uri}</value> </property> <property> <name>hive.metastore.sasl.enabled</name> <value>true</value> </property> <property> <name>hive.metastore.kerberos.principal</name> <value>${hive_metastore_principal}</value> </property> |
- Add also ‐‐skip-dist-cache option to avoid the error:
|
1 |
ERROR org.apache.sqoop.tool.ImportTool - Imported Failed: Can not create a Path from an empty string |
Note: Don’t specify ‐‐target-dir option in this case. Data will be stored in the location to which table is pointing (default is the metastore).
Note: ‐‐hive-overwrite option is not supported with HCatalog. As well as those specifying format:
‐‐as-sequencefile
‐‐as-avrodatafile
‐‐as-parquetfile
Note: Supported options are:
‐‐map-column-hive – to map a database column to HCatalog column type
‐‐hive-partition-key and ‐‐hive-partition-value – for specifying partition key and value, but here only one static partition is allowed in contrast to HCatalog partitioning options!
Here’s how an example Oozie workflow may look like:
|
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 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 |
<workflow-app name="sqoop_with_hcatalog" xmlns="uri:oozie:workflow:0.4"> <global> <job-xml>job-conf.xml</job-xml> </global> <credentials> <credential name="hcat" type="hcat"> <property> <name>hcat.metastore.uri</name> <value>${hcat_metastore_uri}</value> </property> <property> <name>hcat.metastore.principal</name> <value>${hcat_metastore_principal}</value> </property> </credential> </credentials> <start to="import"/> <action name="import" cred="hcat"> <sqoop xmlns="uri:oozie:sqoop-action:0.2"> <job-tracker>${jobTracker}</job-tracker> <name-node>${nameNode}</name-node> <arg>import</arg> <arg>--connect</arg> <arg>jdbc:sqlserver://${sql_server_ip};database=${sql_database}</arg> <arg>--username</arg> <arg>${db_username}</arg> <arg>--password-file</arg> <arg>${password_file_dir}</arg> <arg>--query</arg> <arg>select * from ${schema}.{table_name} where $CONDITIONS </arg> <arg>--hcatalog-database</arg> <arg>{hive_database}</arg> <arg>--hcatalog-table</arg> <arg>{table_name}</arg> <arg>--m</arg> <arg>1</arg> <arg>--skip-dist-cache</arg> <file>sqljdbc4.jar</file> <file>hive-site.xml</file> </sqoop> <ok to="end"/> <error to="kill"/> </action> <kill name="kill"> <message>Action failed, error message[${wf:errorMessage(wf:lastErrorNode())}]</message> </kill> <end name="end"/> </workflow-app> |
Writing into specific partitions
If your table is partitioned and you want to import data only into selected partitions, you can do that with ‐‐hcatalog-partition-keys and ‐‐hcatalog-partition-values options and proper conditions in query where clause. However remember that imported data doesn’t overwrite existing partitions, it gets appended.
You can specify multiple partition levels but you can’t write into more than one specified partition at the time.
Supported combinations:
- Multilevel partitioning:
2 levels of partitions in the table: partition1, partition2
‐‐hcatalog-partition-keys: partition1, partition2
‐‐hcatalog-partition-values: value1, value2
Import condition: where partition1 = value1 and partition2= value2
- Specifying one level partition for 2 levels partitioned table:
2 levels of partitions in the table: partition1, partition2
‐‐hcatalog-partition-keys: partition1
‐‐hcatalog-partition-values: value1
Import condition: where partition1 = value1
Not supported combinations:
- Specifying 2 values for 1 partition:
1 level of partition in the table: partition1
‐‐hcatalog-partition-keys: partition1
‐‐hcatalog-partition-values: value1, value2
Import condition: where partition1 = value1 or parition1 = value2
Although you may achieve same result by just specifying the table and adding where clause to the query:
‐‐hcatalog-table: table_name
Import condition: where partition1 = value1 or parition1 = value2
Here’s how an example Oozie workflow may look like:
|
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 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 |
<workflow-app name="sqoop_with_hcatalog" xmlns="uri:oozie:workflow:0.4"> <global> <job-xml>job-conf.xml</job-xml> </global> <credentials> <credential name="hcat" type="hcat"> <property> <name>hcat.metastore.uri</name> <value>${hcat_metastore_uri}</value> </property> <property> <name>hcat.metastore.principal</name> <value>${hcat_metastore_principal}</value> </property> </credential> </credentials> <start to="import"/> <action name="importt" cred="hcat"> <sqoop xmlns="uri:oozie:sqoop-action:0.2"> <job-tracker>${jobTracker}</job-tracker> <name-node>${nameNode}</name-node> <arg>import</arg> <arg>--connect</arg> <arg>jdbc:sqlserver://${sql_server_ip};database=${sql_database}</arg> <arg>--username</arg> <arg>${db_username}</arg> <arg>--password-file</arg> <arg>${password_file_dir}</arg> <arg>--query</arg> <arg>select * from ${schema}.{table_name} where $CONDITIONS and ${partition1}='${value1}'</arg> <arg>--hcatalog-database</arg> <arg>${hive_database}</arg> <arg>--hcatalog-table</arg> <arg>${partitioned_table}</arg> <arg>--hcatalog-partition-keys</arg> <arg>${partition1}</arg> <arg>--hcatalog-partition-values</arg> <arg>${value1}</arg> <arg>--m</arg> <arg>1</arg> <arg>--skip-dist-cache</arg> <file>sqljdbc4.jar</file> <file>hive-site.xml</file> </sqoop> <ok to="end"/> <error to="kill"/> </action> <kill name="kill"> <message>Action failed, error message[${wf:errorMessage(wf:lastErrorNode())}]</message> </kill> <end name="end"/> </workflow-app> |
Export
Before doing Sqoop export make sure that the table is present on the database side.
Note: By design columns in database table must match the ones in created table! It’s because HCatalog does the column matching by names. If some columns won’t match you’ll get error message:
|
1 |
java.lang.NullPointerException at org.apache.hive.hcatalog.data.schema.HCatSchema.get(HCatSchema.java:105) |
Note: Original database column names are converted to lowercase and ‘-‘ characters are replaced with “_” by Sqoop. This has to be taken into consideration while creating Hive table.
For the Sqoop export action with HCatalog in Oozie, in addition to normal Oozie Sqoop action stuff, you need to add:
- HCatalog credentials
- jdbc jar file
- hive-site.xml file
Your hive-site.xml should contain at least following properties:
|
1 2 3 4 5 6 7 8 9 10 11 12 |
<property> <name>hive.metastore.uri</name> <value>${hive_metastore_uri}</value> </property> <property> <name>hive.metastore.sasl.enabled</name> <value>true</value> </property> <property> <name>hive.metastore.kerberos.principal</name> <value>${hive_metastore_principal}</value> </property> |
- Add also ‐‐skip-dist-cache option to avoid the error:
|
1 |
ERROR org.apache.sqoop.tool.ImportTool - Imported Failed: Can not create a Path from an empty string |
An example Oozie workflow may look like this:
|
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 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 |
<workflow-app name="export_with_hcatalog" xmlns="uri:oozie:workflow:0.4"> <global> <job-xml>job-conf.xml</job-xml> </global> <credentials> <credential name="hcat" type="hcat"> <property> <name>hcat.metastore.uri</name> <value>${hcat_metastore_uri}</value> </property> <property> <name>hcat.metastore.principal</name> <value>${hcat_metastore_principal}</value> </property> </credential> </credentials> <start to="export"/> <action name="export" cred="hcat"> <sqoop xmlns="uri:oozie:sqoop-action:0.2"> <job-tracker>${jobTracker}</job-tracker> <name-node>${nameNode}</name-node> <arg>export</arg> <arg>--connect</arg> <arg>jdbc:sqlserver://${sql_server_ip};database=${sql_database}</arg> <arg>--username</arg> <arg>${db_username}</arg> <arg>--password-file</arg> <arg>${password_file_dir}</arg> <arg>--table</arg> <arg>${db_table_name}</arg> <arg>--hcatalog-database</arg> <arg>${hive_database}</arg> <arg>--hcatalog-table</arg> <arg>${hive_table}</arg> <arg>--m</arg> <arg>1</arg> <arg>--skip-dist-cache</arg> <arg>--</arg> <arg>--schema</arg> <arg>${db_table_schema}</arg> <file>sqljdbc4.jar</file> <file>hive-site.xml</file> </sqoop> <ok to="end"/> <error to="kill"/> </action> <kill name="kill"> <message>Action failed, error message[${wf:errorMessage(wf:lastErrorNode())}]</message> </kill> <end name="end"/> </workflow-app> |