Created on 10-16-201805:14 AM - edited on 01-27-202109:41 AM by VidyaSargur
Short Description: This article targets to describe and demonstrate Apache Hive Warehouse Connector which is a newer generation to read and write data between Apache Spark and Apache Hive.
1. Motivation
Apache Spark and Apache Hive integration has always been an important use case and continues to be so. Both provide their own efficient ways to process data by the use of SQL, and is used for data stored in distributed file systems. Both provide compatibilities for each other. As both systems evolve, it is critical to find a solution that provides the best of both worlds for data processing needs.
In case of Apache Spark, it provides a basic Hive compatibility. It allows an access to tables in Apache Hive and some basic use cases can be achieved by this. However, not all the modern features from Apache Hive are supported, for instance, ACID table in Apache Hive, Ranger integration, Live Long And Process (LLAP), etc. as of Spark 2.
Apache Spark supports a pluggable approach for various data sources and Apache Hive itself can be also considered as one data source. Therefore, this library, Hive Warehouse Connector, was implemented as a data source to overcome the limitations and provide those modern functionalities in Apache Hive to Apache Spark users.
Note: From HDP 3.0, catalogs for Apache Hive and Apache Spark are separated, and they use their own catalog; namely, they are mutually exclusive - Apache Hive catalog can only be accessed by Apache Hive or this library, and Apache Spark catalog can only be accessed by existing APIs in Apache Spark . In other words, some features such as ACID tables or Apache Ranger with Apache Hive table are only available via this library in Apache Spark. Those tables in Hive should not directly be accessible within Apache Spark APIs themselves.
2. Introduction
This library provides both Scala (Java compatible) and Python APIs for:
SQL / DataFrame APIs interacting with both transactional and non-transactional tables in Apache Hive
SQL / DataFrame read support
SQL / DataFrame and Structured Streaming write support
Apache Spark's Apache Arrow integration is fully utilized for vectorized operations, faster and compact data interactions
It is implemented by Data Source V2 which has a columnar format support and various functionalities
For instance, it is able to read an Apache Hive table in Apache Spark as below:
import com.hortonworks.hwc.HiveWarehouseSession
val hive = HiveWarehouseSession.session(spark).build()
val df = hive.executeQuery("SELECT * FROM tableA")
It leverages Apache Hive LLAP and retrieves data from Hive table into Spark DataFrame.
2.2. SQL / DataFrame & Structured Streaming Write Support
Hive Streaming API is used in both batch and streaming write, which Apache Hive introduced to continuously digest data.
Since it is implemented by Data Source V2, it supports a commit protocol and supports atomic write operation.
Native Apache ORC writer is used instead, which has many important fixes in terms of performance and stability.
The code below illustrate writing data from Apache Spark to Apache Hive table in Structured Streaming.
import com.hortonworks.hwc.HiveWarehouseSession
val hive = HiveWarehouseSession.session(spark).build()
df.writeStream
.format(HiveWarehouseSession.STREAM_TO_STREAM)
.option("table", "hwx_table").start()
Internally this fully utilizes Apache Hive’s Streaming API to write Apache’s DataFrame out to Apache Hive’s table.
Note: This does not need Hive LLAP daemons to be running.
3. Prerequisites
In Apache Hive:
‘Interactive Query' (LLAP) should be enabled (see 7. Appendix for Apache Ambari UI)
In Apache Spark:
spark.hadoop.hive.llap.daemon.service.hosts should be set for the application name of the LLAP service since this library utilizes LLAP. For example, @llap0.
HiveServer2's JDBC URL should be specified in spark.sql.hive.hiveserver2.jdbc.url as well as configured in the cluster. For example, jdbc:hive2://localhost:10000. Use the HiveServer2 Interactive JDBC URL, rather than the traditional HiveServer2's JDBC URL.
Make sure spark.datasource.hive.warehouse.load.staging.dir is pointed into a suitable HDFS-compatible staging directory, e.g. /tmp.
Also, ensure spark.datasource.hive.warehouse.metastoreUri is configured properly. For example, thrift://localhost:9083 to indicate Metastore URI.
Note that spark.security.credentials.hiveserver2.enabled should be set to false for YARN client deploy mode, and true for YARN cluster deploy mode (by default). This configuration is required for a Kerberized cluster.
When spark.security.credentials.hiveserver2.enabled is set to false, spark.sql.hive.hiveserver2.jdbc.url.principal can be optionally set if spark.sql.hive.hiveserver2.jdbc.url does not contain principal, for example, hive/_HOST@EXAMPLE.COM.
When spark.security.credentials.hiveserver2.enabled is set to true, spark.sql.hive.hiveserver2.jdbc.url.principal should be set, for example, hive/_HOST@EXAMPLE.COM. Also, spark.sql.hive.hiveserver2.jdbc.url should not contain principal.
4. User Scenarios
In this chapter, HDP 3.0.1.0 with Spark2, Hive and Ranger on a Kerberized cluster is used. Note that If you are running the examples on Kerberized cluster, for instance in HDP, make sure to obtain (or renew) the Kerberos ticket-granting ticket by kinit with an appropriate keytab.
The following scenarios are run by Spark shell as below. For other interaction paradigms, like spark-submit, Livy, Zeppelin etc. please see the subsequent sections for specific setup steps.
This scenario targets to demonstrate a bulk read operation, as a batch job, from Hive table to Spark DataFrame with SQL expression. For this scenario, data from FBI crime rate is used (see 7. Appendix for SQL INSERT queries).
Before we start, we should initialize HiveWarehouseSession instance.
In case of Scala:
import com.hortonworks.hwc.HiveWarehouseSession
val hive = HiveWarehouseSession.session(spark).build()
In case of Python:
from pyspark_llap.sql.session import HiveWarehouseSession
hive = HiveWarehouseSession.session(spark).build()
In this scenarios, it uses hwc_db database. The code below shows the decreasing crime rate (per 100,000 Inhabitants) between 2000 and 2010 in USA:
hive.setDatabase("hwc_db")
hive.table("crimes").filter("year = 2000 OR year = 2010").show()
+----+----------+
|year|crime_rate|
+----+----------+
|2010| 404.5|
|2000| 506.5|
+----+----------+
SQL queries can be also directly executed via executeQuery API, which directly interacts with Hive LLAP daemons.
hive.executeQuery("SELECT avg(crime_rate) AS average_crime_rate FROM crimes").show()
+------------------+
|average_crime_rate|
+------------------+
|456.33500000000004|
+------------------+
As of HDP 2.6, Row/Column Level Access Control for Apache Spark is available that enables fine grained security control leveraged by Apache Ranger. This feature can also be leveraged by this library - Apache Spark APIs do not support this.
For instance, the scenarios below demonstrate row level control with users, billing, and datascience. billing principal can access all rows and columns while datascience principal can access some of filtered and masked data.
Firstly, we destroy existing ticket and obtains a ticket as billing principal. Then, it executes a Spark shell.
For more information, please see 'Introducing Row/ Column Level Access Control for Apache Spark' in 7. Appendix. The Ranger security features previously available in the spark-llap connector are now covered by the Hive Warehouse Connector. Apache Hive and Apache Spark configurations should be followed as described in 3. Prerequisites.
4.3. SQL / DataFrame Write
This scenario targets to demonstrate a bulk write operation, as a batch job, between Apache Hive table and Apache Spark DataFrame with SQL expression.
Here, crimes table (from 4.1 SQL / DataFrame Read) is written into a different Hive table after filtering the data in Spark. The code below writes the crime rate at 2010 into the table created above:
In another terminal, execute below command to insert test data:
$ nc -lk 9999
We create a target table to store the stream data. Note that this table can be created in Hive side as well, for instance, CREATE TABLE hwx_table(value STRING);
Now, the stream data is ready. After that, import the library so that we can easily specify the format.
Before we start, we should initialize HiveWarehouseSession instance.
In case of Scala:
import com.hortonworks.hwc.HiveWarehouseSession
In case of Python:
from pyspark_llap.sql.session import HiveWarehouseSession
Next, it starts the structured streaming job. At the terminal which opened nc -lk 9999 we can insert arbitrary data and when the terminal inserts Hortonworks, it saves the data into the hwx_table table.
Note: There is an issue about interpreting spark.datasource.* configurations into options internally in Apache Spark, which currently makes this library require to set metastoreUri and database options manually. See SPARK-25460 for more details. As soon as this issue is resolved, both metastoreUri and database can be omitted likewise.
After that, type Hortonworks and also arbitrary words in the terminal opening nc -lk 9999
Foo
Bar
Hortonworks
This continuously inserts the word Hortonworks into the table hwx_table .
This library can be interacted with, of course, Apache Spark but also Apache Zeppelin and Apache Livy. This chapter demonstrates entry points of those interaction paradigms in order to leverage this library to utilize provided functionalities.
Note: Local directories are disallowed by default for security reasons. Set the local directory to livy.file.local-dir-whitelist to allow a specific directory to allow to read. To avoid this setting, please use HDFS path after uploading the JAR and zipped file.
If the interpreter is set to YARN client mode (see livy.spark.master and livy.spark.submit.deployMode), livy.spark.security.credentials.hiveserver2.enabled should be set to false as below:
Note: The example above used Kerberized cluster; therefore, --negotiate option is used in curl.
Note: Local directories are disallowed by default for security reasons. Set the local directory to livy.file.local-dir-whitelist to allow a specific directory to allow to read. To avoid this setting, please use HDFS path after uploading the JAR and zipped file.
Note: To use HWC with Livy in a secure cluster please follow the documentation here.
Missing R library support: this library currently supports only Scala, Java, and Python APIs.
Limitations in Apache Arrow integration and Data Source V2 are inherited. See SPARK-22386 for Data Source V2 and SPARK-21187 for Apache Arrow integration in Apache Spark to track ongoing efforts and limitations.
Supported/unsupported data types and their mapping between Apache Hive and Apache Spark can be found at this link.
org.apache.hadoop.yarn.exceptions.YarnException: Failed to submit application_1547158600691_0002 to YARN : org.apache.hadoop.security.AccessControlException: Queue root.default already has 0 applications, cannot accept submission of application: application_1547158600691_0002
at org.apache.hadoop.yarn.client.api.impl.YarnClientImpl.submitApplication(YarnClientImpl.java:304)
at org.apache.spark.deploy.yarn.Client.submitApplication(Client.scala:174)
at org.apache.spark.scheduler.cluster.YarnClientSchedulerBackend.start(YarnClientSchedulerBackend.scala:57)
at org.apache.spark.scheduler.TaskSchedulerImpl.start(TaskSchedulerImpl.scala:164)
at org.apache.spark.SparkContext.<init>(SparkContext.scala:500)
at org.apache.spark.api.java.JavaSparkContext.<init>(JavaSparkContext.scala:58)
at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
at sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62)
at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
at java.lang.reflect.Constructor.newInstance(Constructor.java:423)
at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:247)
at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)
at py4j.Gateway.invoke(Gateway.java:238)
at py4j.commands.ConstructorCommand.invokeConstructor(ConstructorCommand.java:80)
at py4j.commands.ConstructorCommand.execute(ConstructorCommand.java:69)
at py4j.GatewayConnection.run(GatewayConnection.java:238)
at java.lang.Thread.run(Thread.java:748)
19/01/10 22:57:49 WARN YarnSchedulerBackend$YarnSchedulerEndpoint: Attempted to request executors before the AM has registered!
19/01/10 22:57:49 WARN MetricsSystem: Stopping a MetricsSystem that is not running
19/01/10 22:57:49 WARN SparkContext: Another SparkContext is being constructed (or threw an exception in its constructor). This may indicate an error, since only one SparkContext may be running in this JVM (see SPARK-2243). The other SparkContext was created at:
i followed the above steps, and was able to execute commands like ( show databases, show tables) successfully, also created a database from spark-shell and created a table and inserted some data in it, but i am not able to query the data either from the newly created table from spark, nor the tables that already exists in hive, and getting this error
java.lang.AbstractMethodError: Method com/hortonworks/spark/sql/hive/llap/HiveWarehouseDataSourceReader.createBatchDataReaderFactories()Ljava/util/List; is abstract
at com.hortonworks.spark.sql.hive.llap.HiveWarehouseDataSourceReader.createBatchDataReaderFactories(HiveWarehouseDataSourceReader.java)
the commands is as below:
import com.hortonworks.hwc.HiveWarehouseSession
val hive = HiveWarehouseSession.session(spark).build()
