import java.io.File

import org.apache.spark.sql.{Row, SaveMode, SparkSession}

case class Record(key: Int, value: String)

// warehouseLocation points to the default location for managed databases and tables val warehouseLocation = new File(“spark-warehouse”).getAbsolutePath

val spark = SparkSession .builder() .appName(“Spark Hive Example”) .config(“spark.sql.warehouse.dir”, warehouseLocation) .enableHiveSupport() .getOrCreate()

import spark.implicits._ import spark.sql

sql(“CREATE TABLE IF NOT EXISTS src (key INT, value STRING) USING hive”) sql(“LOAD DATA LOCAL INPATH ‘examples/src/main/resources/kv1.txt’ INTO TABLE src”)

// Queries are expressed in HiveQL sql(“SELECT * FROM src”).show() // +—+——-+ // |key| value| // +—+——-+ // |238|val_238| // | 86| val_86| // |311|val_311| // …

// Aggregation queries are also supported. sql(“SELECT COUNT(*) FROM src”).show() // +——–+ // |count(1)| // +——–+ // | 500 | // +——–+

// The results of SQL queries are themselves DataFrames and support all normal functions. val sqlDF = sql(“SELECT key, value FROM src WHERE key < 10 ORDER BY key”)

// The items in DataFrames are of type Row, which allows you to access each column by ordinal. val stringsDS = sqlDF.map { case Row(key: Int, value: String) => s“Key: $key, Value: $value” } stringsDS.show() // +——————–+ // | value| // +——————–+ // |Key: 0, Value: val_0| // |Key: 0, Value: val_0| // |Key: 0, Value: val_0| // …

// You can also use DataFrames to create temporary views within a SparkSession. val recordsDF = spark.createDataFrame((1 to 100).map(i => Record(i, s“val_$i”))) recordsDF.createOrReplaceTempView(“records”)

// Queries can then join DataFrame data with data stored in Hive. sql(“SELECT * FROM records r JOIN src s ON r.key = s.key”).show() // +—+——+—+——+ // |key| value|key| value| // +—+——+—+——+ // | 2| val_2| 2| val_2| // | 4| val_4| 4| val_4| // | 5| val_5| 5| val_5| // …

// Create a Hive managed Parquet table, with HQL syntax instead of the Spark SQL native syntax // USING hive sql(“CREATE TABLE hive_records(key int, value string) STORED AS PARQUET”) // Save DataFrame to the Hive managed table val df = spark.table(“src”) df.write.mode(SaveMode.Overwrite).saveAsTable(“hive_records”) // After insertion, the Hive managed table has data now sql(“SELECT * FROM hive_records”).show() // +—+——-+ // |key| value| // +—+——-+ // |238|val_238| // | 86| val_86| // |311|val_311| // …

// Prepare a Parquet data directory val dataDir = “/tmp/parquet_data” spark.range(10).write.parquet(dataDir) // Create a Hive external Parquet table sql(s“CREATE EXTERNAL TABLE hive_bigints(id bigint) STORED AS PARQUET LOCATION ‘$dataDir’”) // The Hive external table should already have data sql(“SELECT * FROM hive_bigints”).show() // +—+ // | id| // +—+ // | 0| // | 1| // | 2| // … Order may vary, as spark processes the partitions in parallel.

// Turn on flag for Hive Dynamic Partitioning spark.sqlContext.setConf(“hive.exec.dynamic.partition”, “true”) spark.sqlContext.setConf(“hive.exec.dynamic.partition.mode”, “nonstrict”) // Create a Hive partitioned table using DataFrame API df.write.partitionBy(“key”).format(“hive”).saveAsTable(“hive_part_tbl”) // Partitioned column key will be moved to the end of the schema. sql(“SELECT * FROM hive_part_tbl”).show() // +——-+—+ // | value|key| // +——-+—+ // |val_238|238| // | val_86| 86| // |val_311|311| // …

spark.stop()

import java.io.File; import java.io.Serializable; import java.util.ArrayList; import java.util.List;

import org.apache.spark.api.java.function.MapFunction; import org.apache.spark.sql.Dataset; import org.apache.spark.sql.Encoders; import org.apache.spark.sql.Row; import org.apache.spark.sql.SparkSession;

public static class Record implements Serializable { private int key; private String value;

public int getKey() { return key; }

public void setKey(int key) { this.key = key; }

public String getValue() { return value; }

public void setValue(String value) { this.value = value; } }

// warehouseLocation points to the default location for managed databases and tables String warehouseLocation = new File(“spark-warehouse”).getAbsolutePath(); SparkSession spark = SparkSession .builder() .appName(“Java Spark Hive Example”) .config(“spark.sql.warehouse.dir”, warehouseLocation) .enableHiveSupport() .getOrCreate();

spark.sql(“CREATE TABLE IF NOT EXISTS src (key INT, value STRING) USING hive”); spark.sql(“LOAD DATA LOCAL INPATH ‘examples/src/main/resources/kv1.txt’ INTO TABLE src”);

// Queries are expressed in HiveQL spark.sql(“SELECT * FROM src”).show(); // +—+——-+ // |key| value| // +—+——-+ // |238|val_238| // | 86| val_86| // |311|val_311| // …

// Aggregation queries are also supported. spark.sql(“SELECT COUNT(*) FROM src”).show(); // +——–+ // |count(1)| // +——–+ // | 500 | // +——–+

// The results of SQL queries are themselves DataFrames and support all normal functions. Dataset<Row> sqlDF = spark.sql(“SELECT key, value FROM src WHERE key < 10 ORDER BY key”);

// The items in DataFrames are of type Row, which lets you to access each column by ordinal. Dataset<String> stringsDS = sqlDF.map( (MapFunction<Row, String>) row -> “Key: “ + row.get(0) + ”, Value: “ + row.get(1), Encoders.STRING()); stringsDS.show(); // +——————–+ // | value| // +——————–+ // |Key: 0, Value: val_0| // |Key: 0, Value: val_0| // |Key: 0, Value: val_0| // …

// You can also use DataFrames to create temporary views within a SparkSession. List<Record> records = new ArrayList<>(); for (int key = 1; key < 100; key++) { Record record = new Record(); record.setKey(key); record.setValue(“val_” + key); records.add(record); } Dataset<Row> recordsDF = spark.createDataFrame(records, Record.class); recordsDF.createOrReplaceTempView(“records”);

// Queries can then join DataFrames data with data stored in Hive. spark.sql(“SELECT * FROM records r JOIN src s ON r.key = s.key”).show(); // +—+——+—+——+ // |key| value|key| value| // +—+——+—+——+ // | 2| val_2| 2| val_2| // | 2| val_2| 2| val_2| // | 4| val_4| 4| val_4| // …

from os.path import join, abspath

from pyspark.sql import SparkSession from pyspark.sql import Row

# warehouse_location points to the default location for managed databases and tables warehouse_location = abspath(‘spark-warehouse’)

spark = SparkSession \ .builder \ .appName(“Python Spark SQL Hive integration example”) \ .config(“spark.sql.warehouse.dir”, warehouse_location) \ .enableHiveSupport() \ .getOrCreate()

# spark is an existing SparkSession spark.sql(“CREATE TABLE IF NOT EXISTS src (key INT, value STRING) USING hive”) spark.sql(“LOAD DATA LOCAL INPATH ‘examples/src/main/resources/kv1.txt’ INTO TABLE src”)

# Queries are expressed in HiveQL spark.sql(“SELECT * FROM src”).show() # +—+——-+

|key| value|

+—+——-+

|238|val_238|

| 86| val_86|

|311|val_311|

…

</span> # Aggregation queries are also supported. spark.sql(“SELECT COUNT(*) FROM src”).show() # +——–+

|count(1)|

+——–+

| 500 |

+——–+

</span> # The results of SQL queries are themselves DataFrames and support all normal functions. sqlDF = spark.sql(“SELECT key, value FROM src WHERE key < 10 ORDER BY key”)

# The items in DataFrames are of type Row, which allows you to access each column by ordinal. stringsDS = sqlDF.rdd.map(lambda row: “Key: %d, Value: %s” % (row.key, row.value)) for record in stringsDS.collect(): print(record) # Key: 0, Value: val_0

Key: 0, Value: val_0

Key: 0, Value: val_0

…

</span> # You can also use DataFrames to create temporary views within a SparkSession. Record = Row(“key”, “value”) recordsDF = spark.createDataFrame([Record(i, “val_” + str(i)) for i in range(1, 101)]) recordsDF.createOrReplaceTempView(“records”)

# Queries can then join DataFrame data with data stored in Hive. spark.sql(“SELECT * FROM records r JOIN src s ON r.key = s.key”).show() # +—+——+—+——+

|key| value|key| value|

+—+——+—+——+

| 2| val_2| 2| val_2|

| 4| val_4| 4| val_4|

| 5| val_5| 5| val_5|

…

</span>

When working with Hive one must instantiate SparkSession with Hive support. This adds support for finding tables in the MetaStore and writing queries using HiveQL.

# enableHiveSupport defaults to TRUE sparkR.session(enableHiveSupport = TRUE) sql(“CREATE TABLE IF NOT EXISTS src (key INT, value STRING) USING hive”) sql(“LOAD DATA LOCAL INPATH ‘examples/src/main/resources/kv1.txt’ INTO TABLE src”)

</span># Queries can be expressed in HiveQL. results <- collect(sql(“FROM src SELECT key, value”))

</span><div>Find full example code at “examples/src/main/r/RSparkSQLExample.R” in the Spark repo.</div>