Hello Keith,
Please find below Spark HWC Integration java code using Maven.
pom.xml:
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>com.cloudera.spark.hwc</groupId>
<artifactId>SparkCDPHWCExample</artifactId>
<version>1.0.0-SNAPSHOT</version>
<name>SparkCDPHWCExample</name>
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<maven.compiler.source>1.8</maven.compiler.source>
<maven.compiler.target>1.8</maven.compiler.target>
<spark.version>2.4.0.7.1.4.0-203</spark.version>
<hwc.version>1.0.0.7.1.4.0-203</hwc.version>
<spark.scope>provided</spark.scope>
<maven.compiler.plugin.version>3.8.1</maven.compiler.plugin.version>
<maven-shade-plugin.version>3.2.3</maven-shade-plugin.version>
<scala-maven-plugin.version>4.3.1</scala-maven-plugin.version>
<scala.version>2.11</scala.version>
<scala-binary.version>2.12</scala-binary.version>
</properties>
<dependencies>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-sql_2.11</artifactId>
<version>${spark.version}</version>
<scope>${spark.scope}</scope>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-hive_2.11</artifactId>
<version>${spark.version}</version>
<scope>${spark.scope}</scope>
</dependency>
<dependency>
<groupId>com.hortonworks.hive</groupId>
<artifactId>hive-warehouse-connector_2.11</artifactId>
<version>${hwc.version}</version>
<scope>${spark.scope}</scope>
</dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>4.12</version>
<scope>test</scope>
</dependency>
</dependencies>
<build>
<plugins>
<!-- Maven Compiler Plugin -->
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<version>${maven.compiler.plugin.version}</version>
<configuration>
<source>${maven.compiler.source}</source>
<target>${maven.compiler.target}</target>
</configuration>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-shade-plugin</artifactId>
<version>${maven-shade-plugin.version}</version>
<executions>
<!-- Run shade goal on package phase -->
<execution>
<phase>package</phase>
<goals>
<goal>shade</goal>
</goals>
<configuration>
<transformers>
<!-- add Main-Class to manifest file -->
<transformer
implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
<mainClass>com.cloudera.spark.hwc.SparkHWCJavaExample</mainClass>
</transformer>
</transformers>
<!-- Remove signed keys to prevent security exceptions on uber jar -->
<filters>
<filter>
<artifact>*:*</artifact>
<excludes>
<exclude>META-INF/*.SF</exclude>
<exclude>META-INF/*.DSA</exclude>
<exclude>META-INF/*.RSA</exclude>
</excludes>
</filter>
</filters>
</configuration>
</execution>
</executions>
</plugin>
</plugins>
</build>
</project>
Java Code:
Employee.java
package com.cloudera.spark.hwc;
public class Employee {
private long id;
private String name;
private int age;
private double salary;
public Employee() {
}
public Employee(long id, String name, int age, double salary) {
this.id = id;
this.name = name;
this.age = age;
this.salary = salary;
}
public long getId() {
return id;
}
public void setId(long id) {
this.id = id;
}
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
public int getAge() {
return age;
}
public void setAge(int age) {
this.age = age;
}
public double getSalary() {
return salary;
}
public void setSalary(double salary) {
this.salary = salary;
}
}
SparkHWCJavaExample.java
package com.cloudera.spark.hwc;
import org.apache.spark.SparkConf;
import org.apache.spark.sql.*;
import com.hortonworks.hwc.HiveWarehouseSession;
import java.util.ArrayList;
import java.util.List;
public class SparkHWCJavaExample {
private final static String DATABASE_NAME = "hwc_db";
private final static String TABLE_NAME = "employee";
private final static String DATABASE_TABLE_NAME = DATABASE_NAME +"."+TABLE_NAME;
public static void main(String[] args) {
SparkConf sparkConf = new SparkConf().
setSparkHome("Spark CDP HWC Example")
.setIfMissing("spark.master", "local");
SparkSession spark = SparkSession.builder().config(sparkConf)
.enableHiveSupport()
.getOrCreate();
HiveWarehouseSession hive = HiveWarehouseSession.session(spark).build();
// Create a Database
hive.createDatabase(DATABASE_NAME, true);
// Display all databases
hive.showDatabases().show(false);
// Use database
hive.setDatabase(DATABASE_NAME);
// Display all tables
hive.showTables().show(false);
// Drop a table
hive.dropTable(DATABASE_TABLE_NAME, true, true);
// Create a Table
hive.createTable("employee").ifNotExists().column("id", "bigint").column("name", "string").
column("age", "smallint").column("salary", "double").create();
// Insert the data
Encoder<Employee> encoder = Encoders.bean(Employee.class);
List<Employee> employeeList = new ArrayList<>();
employeeList.add(new Employee(1, "Ranga", 32, 245000.30));
employeeList.add(new Employee(2, "Nishanth", 2, 345000.10));
employeeList.add(new Employee(3, "Raja", 32, 245000.86));
employeeList.add(new Employee(4, "Mani", 14, 45000));
Dataset<Employee> employeeDF = spark.createDataset(employeeList, encoder);
employeeDF.write().format("com.hortonworks.spark.sql.hive.llap.HiveWarehouseConnector").
mode("append").option("table", "hwc_db.employee").save();
// Select the data
Dataset<Row> empDF = hive.executeQuery("SELECT * FROM "+DATABASE_TABLE_NAME);
empDF.printSchema();
empDF.show();
// Close the SparkSession
spark.close();
}
}
Run Script:
sudo -u hive spark-submit \
--master yarn \
--deploy-mode client \
--executor-memory 1g \
--num-executors 2 \
--driver-memory 1g \
--jars /opt/cloudera/parcels/CDH/jars/hive-warehouse-connector-assembly-*.jar \
--conf spark.sql.hive.hiveserver2.jdbc.url='jdbc:hive2://localhost:10000' \
--conf spark.sql.hive.hwc.execution.mode=spark \
--conf spark.datasource.hive.warehouse.metastoreUri='thrift://localhost:9083' \
--conf spark.datasource.hive.warehouse.load.staging.dir='/tmp' \
--conf spark.datasource.hive.warehouse.user.name=hive \
--conf spark.datasource.hive.warehouse.password=hive \
--conf spark.datasource.hive.warehouse.smartExecution=false \
--conf spark.datasource.hive.warehouse.read.via.llap=false \
--conf spark.datasource.hive.warehouse.read.jdbc.mode=cluster \
--conf spark.datasource.hive.warehouse.read.mode=DIRECT_READER_V2 \
--conf spark.security.credentials.hiveserver2.enabled=false \
--conf spark.sql.extensions=com.hortonworks.spark.sql.rule.Extensions \
--class com.cloudera.spark.hwc.SparkHWCJavaExample \
/tmp/SparkCDPHWCExample-1.0.0-SNAPSHOT.jar
