花了几天休息的时间整理了这篇文章，就为了让你读完就能深入了解并熟练运用Spark SQL！如果你觉得有用的话请收藏加关注，你的转发和点赞是我最大的动力！原创不易，转载请著名出处！
本文档基于Spark官方网站（spark.apache.org），加上自己的理解和实验编写。

1、概述

Spark SQL是一个用于结构化数据处理的Spark模块。与基本的Spark RDD API不同，Spark SQL提供的接口为Spark提供了关于数据结构和正在执行的计算的更多信息。在内部，Spark SQL使用这些额外的信息来执行额外的优化。有几种与Spark SQL交互的方法，包括SQL和数据集API。在计算结果时，使用相同的执行引擎，而不依赖于用于表示计算的API/语言。这种统一意味着开发人员可以很容易地在不同的api之间来回切换，这些api提供了最自然的方式来表达给定的转换。

1.1、SQL

Spark SQL的一个用途是执行SQL查询。Spark SQL还可以用于从现有Hive中读取数据。在另一种编程语言中运行SQL时，结果将作为数据集/DataFrame返回。您还可以使用命令行或JDBC/ODBC与SQL接口交互。

1.2、Datasets and DataFrames

Dataset是数据的分布式集合。Dataset是Spark 1.6中添加的新接口，它提供了RDDs(强类型、使用强大lambda函数的能力)和Spark SQL优化执行引擎的优点。可以从JVM对象构造数据集，然后使用功能 transformations (map, flatMap, filter, etc.).进行操作。 Dataset API在Scala和Java中可用。Python不支持 Dataset API。但是由于Python的动态特性， Dataset API的许多优点已经可用(例如，您可以按名称自然地访问行。columnname的字段)。R的情况类似。
DataFrame是组织到命名列中的Dataset 。它在概念上等价于关系数据库中的表或R/Python中的数据框架，但是在底层有更丰富的优化。DataFrames可以从许多数据源构建，例如:结构化数据文件、Hive中的表、外部数据库或现有的rdd。DataFrame API在Scala、Java、Python和r中都可以使用。在Scala和Java中，DataFrame 由行Dataset表示。在Scala API中，DataFrame只是Dataset[Row]的类型别名。而在Java API中，用户需要使用Dataset<Row>来表示一个DataFrame。
在整个文档中，我们经常将Scala/Java Datasets of Rows 称为DataFrames。

2、开始入门

目录：

• 起点:SparkSession
• 创建DataFrames
• 非类型化DataSet操作(即DataFrame操作)
• 以编程方式运行SQL查询
• 全局临时视图
• 创建DataSet
• 与RDD进行
  • 使用反射推断模式
  • 以编程方式指定模式
• 聚合
  • 非类型化用户定义的聚合函数
  • 类型安全的用户定义聚合函数

2.1、使用SparkSession开始着手

SparkSession spark = SparkSession
  .builder()
  .appName("Java Spark SQL basic example")
  .config("spark.some.config.option", "some-value")
  .getOrCreate();

Dataset<Row> df = spark.read().json("examples/src/main/resources/people.json");

// Displays the content of the DataFrame to stdout
df.show();
// +----+-------+
// | age|   name|
// +----+-------+
// |null|Michael|
// |  30|   Andy|
// |  19| Justin|
// +----+-------+

在上例中，使用sparksession.read.json方法读取一个java文件，并返回一个dataset对象。通过调用dataset的方法可以进行数据的操作。

2.2、非类型化DataSet操作(即DataFrame操作)

DataFrames为Scala、Java、Python和R中的结构化数据操作提供了一种特定于领域的语言。
如上所述，在Spark 2.0中，DataFrames只是Scala和Java API中的行DataSet。与强类型Scala/Java数据集提供的“类型化转换”相比，这些操作也称为“非类型化转换”。
以下是一些使用数据集进行结构化数据处理的基本例子:

// col("...") is preferable to df.col("...")
import static org.apache.spark.sql.functions.col;

// Print the schema in a tree format
df.printSchema();
// root
// |-- age: long (nullable = true)
// |-- name: string (nullable = true)

// Select only the "name" column
df.select("name").show();
// +-------+
// |   name|
// +-------+
// |Michael|
// |   Andy|
// | Justin|
// +-------+

// Select everybody, but increment the age by 1
df.select(col("name"), col("age").plus(1)).show();
// +-------+---------+
// |   name|(age + 1)|
// +-------+---------+
// |Michael|     null|
// |   Andy|       31|
// | Justin|       20|
// +-------+---------+

// Select people older than 21
df.filter(col("age").gt(21)).show();
// +---+----+
// |age|name|
// +---+----+
// | 30|Andy|
// +---+----+

// Count people by age
df.groupBy("age").count().show();
// +----+-----+
// | age|count|
// +----+-----+
// |  19|    1|
// |null|    1|
// |  30|    1|
// +----+-----+

除了简单的列引用和表达式外，数据集还具有丰富的函数库，包括字符串操作、日期算法、常见的数学操作等等。完整的列表可以在DataFrame函数引用中获得。（http://spark.apache.org/docs/latest/api/java/org/apache/spark/sql/functions.html）

2.3、以编程方式运行SQL查询

SparkSession上的sql函数允许应用程序以编程方式运行sql查询，并以Dataset<Row>的形式返回结果。

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;

// Register the DataFrame as a SQL temporary view
df.createOrReplaceTempView("people");

Dataset<Row> sqlDF = spark.sql("SELECT * FROM people");
sqlDF.show();
// +----+-------+
// | age|   name|
// +----+-------+
// |null|Michael|
// |  30|   Andy|
// |  19| Justin|
// +----+-------+

2.4、全局临时视图

Spark SQL中的临时视图是会话范围的，如果创建它的会话终止，它将消失。如果您希望所有会话之间共享一个临时视图，并在Spark应用程序终止之前保持活动状态，那么您可以创建一个全局临时视图。全局临时视图被绑定到系统保存的数据库global_temp，我们必须使用限定名来引用它，例如，SELECT * FROM global_temp.view1.

// Register the DataFrame as a global temporary view
df.createGlobalTempView("people");

// Global temporary view is tied to a system preserved database `global_temp`
spark.sql("SELECT * FROM global_temp.people").show();
// +----+-------+
// | age|   name|
// +----+-------+
// |null|Michael|
// |  30|   Andy|
// |  19| Justin|
// +----+-------+

// Global temporary view is cross-session
spark.newSession().sql("SELECT * FROM global_temp.people").show();
// +----+-------+
// | age|   name|
// +----+-------+
// |null|Michael|
// |  30|   Andy|
// |  19| Justin|
// +----+-------+

2.5、创建Datasets

Dataset类似于RDDs，但是，它们不使用Java序列化或Kryo，而是使用专门的编码器序列化对象，以便通过网络进行处理或传输。虽然编码器和标准序列化都负责将对象转换成字节，但编码器是动态生成的代码，使用的格式允许Spark执行许多操作，如过滤、排序和哈希，而无需将字节反序列化回对象。

import java.util.Arrays;
import java.util.Collections;
import java.io.Serializable;

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

public static class Person implements Serializable {
  private String name;
  private int age;

  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;
  }
}

// Create an instance of a Bean class
Person person = new Person();
person.setName("Andy");
person.setAge(32);

// Encoders are created for Java beans
Encoder<Person> personEncoder = Encoders.bean(Person.class);
Dataset<Person> javaBeanDS = spark.createDataset(
  Collections.singletonList(person),
  personEncoder
);
javaBeanDS.show();
// +---+----+
// |age|name|
// +---+----+
// | 32|Andy|
// +---+----+

// Encoders for most common types are provided in class Encoders
Encoder<Integer> integerEncoder = Encoders.INT();
Dataset<Integer> primitiveDS = spark.createDataset(Arrays.asList(1, 2, 3), integerEncoder);
Dataset<Integer> transformedDS = primitiveDS.map(
    (MapFunction<Integer, Integer>) value -> value + 1,
    integerEncoder);
transformedDS.collect(); // Returns [2, 3, 4]

// DataFrames can be converted to a Dataset by providing a class. Mapping based on name
String path = "examples/src/main/resources/people.json";
Dataset<Person> peopleDS = spark.read().json(path).as(personEncoder);
peopleDS.show();
// +----+-------+
// | age|   name|
// +----+-------+
// |null|Michael|
// |  30|   Andy|
// |  19| Justin|
// +----+-------+

2.6、进行RDD操作

Spark SQL支持两种不同的方法来将现有的RDDs转换为数据集。第一种方法使用反射来推断包含特定类型对象的RDD的模式。这种基于反射的方法可以生成更简洁的代码，并且当您在编写Spark应用程序时已经知道模式时，这种方法可以很好地工作。
创建数据集的第二种方法是通过编程接口，该接口允许您构造模式，然后将其应用于现有的RDD。虽然这个方法比较冗长，但它允许您在运行时才知道列及其类型时构造数据集。

• 使用反射推断模式

import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.MapFunction;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.Encoder;
import org.apache.spark.sql.Encoders;

// Create an RDD of Person objects from a text file
JavaRDD<Person> peopleRDD = spark.read()
  .textFile("examples/src/main/resources/people.txt")
  .javaRDD()
  .map(line -> {
    String[] parts = line.split(",");
    Person person = new Person();
    person.setName(parts[0]);
    person.setAge(Integer.parseInt(parts[1].trim()));
    return person;
  });

// Apply a schema to an RDD of JavaBeans to get a DataFrame
Dataset<Row> peopleDF = spark.createDataFrame(peopleRDD, Person.class);
// Register the DataFrame as a temporary view
peopleDF.createOrReplaceTempView("people");

// SQL statements can be run by using the sql methods provided by spark
Dataset<Row> teenagersDF = spark.sql("SELECT name FROM people WHERE age BETWEEN 13 AND 19");

// The columns of a row in the result can be accessed by field index
Encoder<String> stringEncoder = Encoders.STRING();
Dataset<String> teenagerNamesByIndexDF = teenagersDF.map(
    (MapFunction<Row, String>) row -> "Name: " + row.getString(0),
    stringEncoder);
teenagerNamesByIndexDF.show();
// +------------+
// |       value|
// +------------+
// |Name: Justin|
// +------------+

// or by field name
Dataset<String> teenagerNamesByFieldDF = teenagersDF.map(
    (MapFunction<Row, String>) row -> "Name: " + row.<String>getAs("name"),
    stringEncoder);
teenagerNamesByFieldDF.show();
// +------------+
// |       value|
// +------------+
// |Name: Justin|
// +------------+

• 以编程方式指定模式
  当不能预先定义JavaBean类时(例如，记录的结构编码在字符串中，或者解析文本数据集，并且针对不同的用户以不同的方式投影字段)，可以通过三个步骤以编程方式创建数据集。
  1.从原始RDD中创建行RDD;
  2.创建由一个StructType表示的模式，该结构与步骤1中创建的RDD中的行结构匹配。
  3.通过SparkSession提供的createDataFrame方法将模式应用到行RDD。

import java.util.ArrayList;
import java.util.List;

import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.function.Function;

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;

import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;

// Create an RDD
JavaRDD<String> peopleRDD = spark.sparkContext()
  .textFile("examples/src/main/resources/people.txt", 1)
  .toJavaRDD();

// The schema is encoded in a string
String schemaString = "name age";

// Generate the schema based on the string of schema
List<StructField> fields = new ArrayList<>();
for (String fieldName : schemaString.split(" ")) {
  StructField field = DataTypes.createStructField(fieldName, DataTypes.StringType, true);
  fields.add(field);
}
StructType schema = DataTypes.createStructType(fields);

// Convert records of the RDD (people) to Rows
JavaRDD<Row> rowRDD = peopleRDD.map((Function<String, Row>) record -> {
  String[] attributes = record.split(",");
  return RowFactory.create(attributes[0], attributes[1].trim());
});

// Apply the schema to the RDD
Dataset<Row> peopleDataFrame = spark.createDataFrame(rowRDD, schema);

// Creates a temporary view using the DataFrame
peopleDataFrame.createOrReplaceTempView("people");

// SQL can be run over a temporary view created using DataFrames
Dataset<Row> results = spark.sql("SELECT name FROM people");

// The results of SQL queries are DataFrames and support all the normal RDD operations
// The columns of a row in the result can be accessed by field index or by field name
Dataset<String> namesDS = results.map(
    (MapFunction<Row, String>) row -> "Name: " + row.getString(0),
    Encoders.STRING());
namesDS.show();
// +-------------+
// |        value|
// +-------------+
// |Name: Michael|
// |   Name: Andy|
// | Name: Justin|
// +-------------+

2.7、聚集操作

内置的DataFrames函数提供常见的聚合，如count()、countDistinct()、avg()、max()、min()等。此外，用户不受预定义聚合函数的限制，可以创建自己的聚合函数。

• 非类型化用户定义的聚合函数
  用户必须扩展UserDefinedAggregateFunction抽象类来实现自定义非类型化聚合函数。例如，用户定义的平均值可以如下所示:

import java.util.ArrayList;
import java.util.List;

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.expressions.MutableAggregationBuffer;
import org.apache.spark.sql.expressions.UserDefinedAggregateFunction;
import org.apache.spark.sql.types.DataType;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;

public static class MyAverage extends UserDefinedAggregateFunction {

  private StructType inputSchema;
  private StructType bufferSchema;

  public MyAverage() {
    List<StructField> inputFields = new ArrayList<>();
    inputFields.add(DataTypes.createStructField("inputColumn", DataTypes.LongType, true));
    inputSchema = DataTypes.createStructType(inputFields);

    List<StructField> bufferFields = new ArrayList<>();
    bufferFields.add(DataTypes.createStructField("sum", DataTypes.LongType, true));
    bufferFields.add(DataTypes.createStructField("count", DataTypes.LongType, true));
    bufferSchema = DataTypes.createStructType(bufferFields);
  }
  // Data types of input arguments of this aggregate function
  public StructType inputSchema() {
    return inputSchema;
  }
  // Data types of values in the aggregation buffer
  public StructType bufferSchema() {
    return bufferSchema;
  }
  // The data type of the returned value
  public DataType dataType() {
    return DataTypes.DoubleType;
  }
  // Whether this function always returns the same output on the identical input
  public boolean deterministic() {
    return true;
  }
  // Initializes the given aggregation buffer. The buffer itself is a `Row` that in addition to
  // standard methods like retrieving a value at an index (e.g., get(), getBoolean()), provides
  // the opportunity to update its values. Note that arrays and maps inside the buffer are still
  // immutable.
  public void initialize(MutableAggregationBuffer buffer) {
    buffer.update(0, 0L);
    buffer.update(1, 0L);
  }
  // Updates the given aggregation buffer `buffer` with new input data from `input`
  public void update(MutableAggregationBuffer buffer, Row input) {
    if (!input.isNullAt(0)) {
      long updatedSum = buffer.getLong(0) + input.getLong(0);
      long updatedCount = buffer.getLong(1) + 1;
      buffer.update(0, updatedSum);
      buffer.update(1, updatedCount);
    }
  }
  // Merges two aggregation buffers and stores the updated buffer values back to `buffer1`
  public void merge(MutableAggregationBuffer buffer1, Row buffer2) {
    long mergedSum = buffer1.getLong(0) + buffer2.getLong(0);
    long mergedCount = buffer1.getLong(1) + buffer2.getLong(1);
    buffer1.update(0, mergedSum);
    buffer1.update(1, mergedCount);
  }
  // Calculates the final result
  public Double evaluate(Row buffer) {
    return ((double) buffer.getLong(0)) / buffer.getLong(1);
  }
}

// Register the function to access it
spark.udf().register("myAverage", new MyAverage());

Dataset<Row> df = spark.read().json("examples/src/main/resources/employees.json");
df.createOrReplaceTempView("employees");
df.show();
// +-------+------+
// |   name|salary|
// +-------+------+
// |Michael|  3000|
// |   Andy|  4500|
// | Justin|  3500|
// |  Berta|  4000|
// +-------+------+

Dataset<Row> result = spark.sql("SELECT myAverage(salary) as average_salary FROM employees");
result.show();
// +--------------+
// |average_salary|
// +--------------+
// |        3750.0|
// +--------------+

• 类型安全的用户定义聚合函数
  强类型数据集的用户定义聚合围绕聚合器抽象类。例如，类型安全的用户定义平均值可以如下所示:

import java.io.Serializable;

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Encoder;
import org.apache.spark.sql.Encoders;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.TypedColumn;
import org.apache.spark.sql.expressions.Aggregator;

public static class Employee implements Serializable {
  private String name;
  private long salary;

  // Constructors, getters, setters...

}

public static class Average implements Serializable  {
  private long sum;
  private long count;

  // Constructors, getters, setters...

}

public static class MyAverage extends Aggregator<Employee, Average, Double> {
  // A zero value for this aggregation. Should satisfy the property that any b + zero = b
  public Average zero() {
    return new Average(0L, 0L);
  }
  // Combine two values to produce a new value. For performance, the function may modify `buffer`
  // and return it instead of constructing a new object
  public Average reduce(Average buffer, Employee employee) {
    long newSum = buffer.getSum() + employee.getSalary();
    long newCount = buffer.getCount() + 1;
    buffer.setSum(newSum);
    buffer.setCount(newCount);
    return buffer;
  }
  // Merge two intermediate values
  public Average merge(Average b1, Average b2) {
    long mergedSum = b1.getSum() + b2.getSum();
    long mergedCount = b1.getCount() + b2.getCount();
    b1.setSum(mergedSum);
    b1.setCount(mergedCount);
    return b1;
  }
  // Transform the output of the reduction
  public Double finish(Average reduction) {
    return ((double) reduction.getSum()) / reduction.getCount();
  }
  // Specifies the Encoder for the intermediate value type
  public Encoder<Average> bufferEncoder() {
    return Encoders.bean(Average.class);
  }
  // Specifies the Encoder for the final output value type
  public Encoder<Double> outputEncoder() {
    return Encoders.DOUBLE();
  }
}

Encoder<Employee> employeeEncoder = Encoders.bean(Employee.class);
String path = "examples/src/main/resources/employees.json";
Dataset<Employee> ds = spark.read().json(path).as(employeeEncoder);
ds.show();
// +-------+------+
// |   name|salary|
// +-------+------+
// |Michael|  3000|
// |   Andy|  4500|
// | Justin|  3500|
// |  Berta|  4000|
// +-------+------+

MyAverage myAverage = new MyAverage();
// Convert the function to a `TypedColumn` and give it a name
TypedColumn<Employee, Double> averageSalary = myAverage.toColumn().name("average_salary");
Dataset<Double> result = ds.select(averageSalary);
result.show();
// +--------------+
// |average_salary|
// +--------------+
// |        3750.0|
// +--------------+

3、数据源

Spark SQL支持通过DataFrame接口对各种数据源进行操作。DataFrame可以使用关系转换操作，也可以用来创建临时视图。将DataFrame注册为临时视图允许对其数据运行SQL查询。本节描述使用Spark数据源加载和保存数据的一般方法，然后讨论内置数据源可用的特定选项。

3.1、通用的加载/保存功能

在最简单的形式中，默认数据源为parquet(除非将spark.sql.sources.default手动指定选项)将用于所有操作。

Dataset<Row> usersDF = spark.read().load("examples/src/main/resources/users.parquet");
usersDF.select("name", "favorite_color").write().save("namesAndFavColors.parquet");

• 手动指定选项
  您还可以手动指定要使用的数据源以及希望传递给数据源的任何额外选项。数据源由其完全限定的名称(即但是对于内置的源代码，您也可以使用它们的短名称(json、parquet、jdbc、orc、libsvm、csv、text)。从任何数据源类型加载的DataFrames都可以使用此语法转换为其他类型。

要加载JSON文件，可以使用:

Dataset<Row> peopleDF =
  spark.read().format("json").load("examples/src/main/resources/people.json");
peopleDF.select("name", "age").write().format("parquet").save("namesAndAges.parquet");

要加载CSV文件，你可以使用:

Dataset<Row> peopleDFCsv = spark.read().format("csv")
  .option("sep", ";")
  .option("inferSchema", "true")
  .option("header", "true")
  .load("examples/src/main/resources/people.csv");

额外的选项也在写操作期间使用。例如，可以控制ORC数据源的Bloom过滤器和字典编码。下面的ORC示例将在 favorite_color上创建Bloom过滤器，并对name and favorite_color使用字典编码。对于parquet，也存在parquet.enable.dictionary选项。要了解更多关于额外ORC/Parquet选项的详细信息，请访问官方的Apache ORC/Parquet网站。

usersDF.write.format("orc")
  .option("orc.bloom.filter.columns", "favorite_color")
  .option("orc.dictionary.key.threshold", "1.0")
  .save("users_with_options.orc")

• 直接对文件运行SQL
  您也可以使用SQL直接查询文件，而不是使用read-api将文件加载到数据框中并进行查询。

Dataset<Row> sqlDF =
  spark.sql("SELECT * FROM parquet.`examples/src/main/resources/users.parquet`");

• 保存模式
  保存操作可以选择采用保存模式，该模式指定如何处理现有数据（如果存在）。重要的是要认识到这些保存模式不使用任何锁定，也不是原子的。此外，在执行覆盖时，数据将在写入新数据之前被删除。

  
  image.png
• 保存到持久表
  也可以使用saveastable命令将数据帧作为持久表保存到hive元存储中。请注意，使用此功能不需要现有的配置单元部署。Spark将为您创建一个默认的本地Hive元存储（使用Derby）。与creatorreplacetempview命令不同，saveastable将具体化数据帧的内容，并创建指向配置单元元存储中数据的指针。即使在spark程序重新启动之后，只要您保持与同一个元存储的连接，持久表仍然存在。通过在具有表名称的SparkSession上调用Table方法，可以创建持久表的数据帧。

对于基于文件的数据源，例如文本、拼花、JSON等，可以通过路径选项指定自定义表路径，例如df.write.option（“path”，“/some/path”）.saveastable（“t”）。删除表时，自定义表路径将不会被删除，表数据仍然存在。如果没有指定自定义表路径，spark会将数据写入仓库目录下的默认表路径。删除表时，默认的表路径也将被删除。

从spark 2.1开始，持久数据源表在hive元存储中存储了每个分区的元数据。这带来了几个好处：

由于元存储只能返回查询所需的分区，因此不再需要查找表的第一个查询上的所有分区。
配置单元DDL，如改变表分区…集合位置现在可用于使用数据源API创建的表。
请注意，在创建外部数据源表（具有路径选项的表）时，默认情况下不会收集分区信息。要同步元存储中的分区信息，可以调用MSCK REPAIR TABLE.

• 包装、分类和分区
  对于基于文件的数据源，也可以对输出进行存储、排序或分区。存储桶和排序仅适用于持久表：

peopleDF.write().bucketBy(42, "name").sortBy("age").saveAsTable("people_bucketed");

而在使用数据集API时，分区可以与save和saveastable一起使用。

usersDF
  .write()
  .partitionBy("favorite_color")
  .format("parquet")
  .save("namesPartByColor.parquet");

对于单个表，可以同时使用分区和bucketing：

peopleDF
  .write()
  .partitionBy("favorite_color")
  .bucketBy(42, "name")
  .saveAsTable("people_partitioned_bucketed");

PartitionBy创建一个目录结构，如分区发现部分所述。因此，它对于具有高基数的列的适用性有限。与此相反，bucketby将数据分布在固定数量的存储桶中，并且可以在多个唯一值未绑定时使用。

3.2、Parquet文件

parquet是许多其他数据处理系统支持的柱状格式。Spark SQL支持读取和写入自动保留原始数据模式的拼花文件。写入拼花文件时，由于兼容性原因，所有列都自动转换为可以为空。

• 以编程方式加载数据
  使用上述示例中的数据：

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

Dataset<Row> peopleDF = spark.read().json("examples/src/main/resources/people.json");

// DataFrames can be saved as Parquet files, maintaining the schema information
peopleDF.write().parquet("people.parquet");

// Read in the Parquet file created above.
// Parquet files are self-describing so the schema is preserved
// The result of loading a parquet file is also a DataFrame
Dataset<Row> parquetFileDF = spark.read().parquet("people.parquet");

// Parquet files can also be used to create a temporary view and then used in SQL statements
parquetFileDF.createOrReplaceTempView("parquetFile");
Dataset<Row> namesDF = spark.sql("SELECT name FROM parquetFile WHERE age BETWEEN 13 AND 19");
Dataset<String> namesDS = namesDF.map(
    (MapFunction<Row, String>) row -> "Name: " + row.getString(0),
    Encoders.STRING());
namesDS.show();
// +------------+
// |       value|
// +------------+
// |Name: Justin|
// +------------+

• 分区发现
  表分区是类似hive的系统中常用的优化方法。在分区表中，数据通常存储在不同的目录中，分区列值编码在每个分区目录的路径中。所有内置文件源（包括text/csv/json/orc/parquet）都能够自动发现和推断分区信息。例如，我们可以使用以下目录结构将以前使用过的所有人口数据存储到一个分区表中，并使用两个额外的列（gender和country）作为分区列：

path
└── to
    └── table
        ├── gender=male
        │   ├── ...
        │   │
        │   ├── country=US
        │   │   └── data.parquet
        │   ├── country=CN
        │   │   └── data.parquet
        │   └── ...
        └── gender=female
            ├── ...
            │
            ├── country=US
            │   └── data.parquet
            ├── country=CN
            │   └── data.parquet
            └── ...

通过将path/to/table传递给sparksession.read.parquet或sparksession.read.load，spark sql将自动从路径中提取分区信息。现在，返回的数据帧的模式变为：

root
|-- name: string (nullable = true)
|-- age: long (nullable = true)
|-- gender: string (nullable = true)
|-- country: string (nullable = true)

请注意，分区列的数据类型是自动推断的。目前，支持数字数据类型、日期、时间戳和字符串类型。有时用户可能不想自动推断分区列的数据类型。对于这些用例，可以通过spark.sql.sources.partitionColumnTypeInterrusion.enabled配置自动类型推断，默认值为true。当类型推断被禁用时，字符串类型将用于分区列。

从spark 1.6.0开始，默认情况下，分区发现仅查找给定路径下的分区。对于上面的示例，如果用户将path/to/table/gender=male传递给sparksession.read.parquet或sparksession.read.load，则gender不会被视为分区列。如果用户需要指定分区发现应该使用的基本路径，则可以在数据源选项中设置基本路径。例如，当path/to/table/gender=male是数据的路径，并且用户将basepath设置为path/to/table/时，gender将是分区列。

• 模式合并
  与Protocol Buffer, Avro, and Thrift一样，parquet也支持模式演化。用户可以从一个简单的模式开始，并根据需要逐步向该模式添加更多列。这样，用户最终可能会得到多个具有不同但相互兼容模式的parquet文件。parquet数据源现在能够自动检测到这种情况并合并所有这些文件的模式。

由于模式合并是一个相对昂贵的操作，并且在大多数情况下不是必需的，所以我们默认从1.5.0开始将其关闭。您可以通过

在读取拼花文件时（如下面的示例所示），将数据源选项mergeschema设置为true，或者
将全局SQL选项spark.sql.parquet.mergeschema设置为true。

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

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;

public static class Square implements Serializable {
  private int value;
  private int square;

  // Getters and setters...

}

public static class Cube implements Serializable {
  private int value;
  private int cube;

  // Getters and setters...

}

List<Square> squares = new ArrayList<>();
for (int value = 1; value <= 5; value++) {
  Square square = new Square();
  square.setValue(value);
  square.setSquare(value * value);
  squares.add(square);
}

// Create a simple DataFrame, store into a partition directory
Dataset<Row> squaresDF = spark.createDataFrame(squares, Square.class);
squaresDF.write().parquet("data/test_table/key=1");

List<Cube> cubes = new ArrayList<>();
for (int value = 6; value <= 10; value++) {
  Cube cube = new Cube();
  cube.setValue(value);
  cube.setCube(value * value * value);
  cubes.add(cube);
}

// Create another DataFrame in a new partition directory,
// adding a new column and dropping an existing column
Dataset<Row> cubesDF = spark.createDataFrame(cubes, Cube.class);
cubesDF.write().parquet("data/test_table/key=2");

// Read the partitioned table
Dataset<Row> mergedDF = spark.read().option("mergeSchema", true).parquet("data/test_table");
mergedDF.printSchema();

// The final schema consists of all 3 columns in the Parquet files together
// with the partitioning column appeared in the partition directory paths
// root
//  |-- value: int (nullable = true)
//  |-- square: int (nullable = true)
//  |-- cube: int (nullable = true)
//  |-- key: int (nullable = true)

• Hive/Parquet Schema Reconciliation
  从表模式处理的角度来看，hive和parquet有两个关键区别。
  1.Hive不区分大小写，而parquet不区分大小写
  2.hive认为所有的列都可以为空，而拼花地板中的空性是很重要的。

因此，在将Hive元存储parquet表转换为Spark SQL parquet表时，必须将Hive元存储模式与parquet模式协调一致。调节规则为：

1. 两个架构中具有相同名称的字段必须具有相同的数据类型，而不管是否可以为空。协调字段应具有parquet侧的数据类型，以便考虑为空性。
2. 已协调的架构正好包含在配置单元元存储架构中定义的字段。
   • 只出现在parquet图模式中的任何字段都将在已协调的模式中删除。
   • 只出现在配置单元元存储架构中的任何字段都将作为可空字段添加到已协调的架构中。

• Metadata Refreshing
  Spark SQL缓存parquet元数据以获得更好的性能。当启用hive metastore parquet表转换时，也会缓存这些转换表的元数据。如果这些表是由配置单元或其他外部工具更新的，则需要手动刷新它们以确保元数据的一致性。

// spark is an existing SparkSession
spark.catalog().refreshTable("my_table");

• Configuration
  parquet的配置可以使用sparksession上的setconf方法完成，也可以使用sql运行set key=value命令。
  image.png

3.3 ORC文件

自Spark 2.3以来，Spark支持一种针对ORC文件的带有新ORC文件格式的矢量化ORC读卡器。为此，新添加了以下配置。当spark.sql.orc.impl设置为native并且spark.sql.orc.enableVectorizedReader设置为true时，矢量化读卡器用于本机ORC表（例如，使用ORC子句创建的表）。对于hive orc serde表（例如，使用hive选项（fileformat'orc'）的子句创建的表），当spark.sql.hive.convertmetastoreorc也设置为true时，将使用矢量化读卡器。

image.png

3.4 JSON文件

spark sql可以自动推断JSON数据集的模式，并将其作为DataSet<ROW>加载。可以在DataSet<string>或JSON文件上使用sparksession.read（）.json（）完成此转换。

请注意，作为JSON文件提供的文件不是典型的JSON文件。每一行必须包含一个独立的、自包含的有效JSON对象。有关详细信息，请参阅(http://jsonlines.org/)

对于常规的多行JSON文件，将multiLine选项设置为true。

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;

// A JSON dataset is pointed to by path.
// The path can be either a single text file or a directory storing text files
Dataset<Row> people = spark.read().json("examples/src/main/resources/people.json");

// The inferred schema can be visualized using the printSchema() method
people.printSchema();
// root
//  |-- age: long (nullable = true)
//  |-- name: string (nullable = true)

// Creates a temporary view using the DataFrame
people.createOrReplaceTempView("people");

// SQL statements can be run by using the sql methods provided by spark
Dataset<Row> namesDF = spark.sql("SELECT name FROM people WHERE age BETWEEN 13 AND 19");
namesDF.show();
// +------+
// |  name|
// +------+
// |Justin|
// +------+

// Alternatively, a DataFrame can be created for a JSON dataset represented by
// a Dataset<String> storing one JSON object per string.
List<String> jsonData = Arrays.asList(
        "{\"name\":\"Yin\",\"address\":{\"city\":\"Columbus\",\"state\":\"Ohio\"}}");
Dataset<String> anotherPeopleDataset = spark.createDataset(jsonData, Encoders.STRING());
Dataset<Row> anotherPeople = spark.read().json(anotherPeopleDataset);
anotherPeople.show();
// +---------------+----+
// |        address|name|
// +---------------+----+
// |[Columbus,Ohio]| Yin|
// +---------------+----+

3.5 Hive Tables

Spark SQL还支持读取和写入存储在Apache配置单元中的数据。但是，由于hive有大量依赖项，因此这些依赖项不包括在默认的spark分布中。如果在类路径上可以找到hive依赖项，spark将自动加载它们。请注意，这些配置单元依赖项也必须存在于所有工作节点上，因为它们需要访问配置单元序列化和反序列化库（SERDE），以便访问存储在配置单元中的数据。

配置配置单元是通过将hive-site.xml、core-site.xml（用于安全配置）和hdfs-site.xml（用于hdfs配置）文件放在conf/中完成的。

使用hive时，必须使用hive支持实例化sparksession，包括与持久hive元存储的连接、对hive serdes的支持以及hive用户定义函数。没有现有配置单元部署的用户仍然可以启用配置单元支持。当hive-site.xml未配置时，上下文会自动在当前目录中创建metastore_db，并创建由spark.sql.warehouse.dir配置的目录，该目录默认为当前目录中启动spark应用程序的目录spark warehouse。注意，hive-site.xml中的hive.metastore.warehouse.dir属性由于spark 2.0.0而被弃用。相反，使用spark.sql.warehouse.dir指定数据库在仓库中的默认位置。您可能需要向启动Spark应用程序的用户授予写权限。

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|
// ...

• 指定hive表的存储格式
  当您创建一个hive表时，您需要定义这个表应该如何从文件系统读/写数据，即“输入格式”和“输出格式”。您还需要定义此表应如何将数据反序列化为行，或将行序列化为数据，即“serde”。以下选项可用于指定存储格式（“serde”、“input format”、“output format”），例如，使用hive选项（fileformat“parquet”）创建table src（id int）。默认情况下，我们将以纯文本形式读取表文件。注意，在创建表时还不支持Hive存储处理程序，您可以使用Hive端的存储处理程序创建表，并使用Spark SQL来读取它。

  
  image.png

• 与不同版本的 Hive Metastore交互
  Spark SQL的Hive支持中最重要的部分之一是与Hive元存储的交互，这使得Spark SQL能够访问Hive表的元数据。从spark 1.4.0开始，可以使用下面描述的配置，使用spark SQL的单个二进制构建来查询不同版本的hive元存储。注意，与用于与元存储对话的Hive版本无关，内部Spark SQL将根据Hive 1.2.1编译，并将这些类用于内部执行（serdes、udfs、udafs等）。
  以下选项可用于配置用于检索元数据的配置单元版本：

image.png

3.6 JDBC To Other Databases

Spark SQL还包括一个数据源，可以使用JDBC从其他数据库中读取数据。与使用JDBCRDD相比，应该优先使用此功能。这是因为结果以DataFrame 的形式返回，并且可以在Spark SQL中轻松地处理它们，或者与其他数据源联接。JDBC数据源也更容易从Java或Python中使用，因为它不需要用户提供CLSASTG。（请注意，这与Spark SQL JDBC服务器不同，后者允许其他应用程序使用Spark SQL运行查询）。

要开始，您需要在spark类路径上包含特定数据库的JDBC驱动程序。例如，要从spark shell连接到postgres，您将运行以下命令：

bin/spark-shell --driver-class-path postgresql-9.4.1207.jar --jars postgresql-9.4.1207.jar

可以使用数据源API将远程数据库中的表作为数据帧或Spark SQL临时视图加载。用户可以在数据源选项中指定JDBC连接属性。用户和密码通常作为登录到数据源的连接属性提供。除了连接属性外，Spark还支持以下不区分大小写的选项：

image.png
image.png
image.png

// Note: JDBC loading and saving can be achieved via either the load/save or jdbc methods
// Loading data from a JDBC source
Dataset<Row> jdbcDF = spark.read()
  .format("jdbc")
  .option("url", "jdbc:postgresql:dbserver")
  .option("dbtable", "schema.tablename")
  .option("user", "username")
  .option("password", "password")
  .load();

Properties connectionProperties = new Properties();
connectionProperties.put("user", "username");
connectionProperties.put("password", "password");
Dataset<Row> jdbcDF2 = spark.read()
  .jdbc("jdbc:postgresql:dbserver", "schema.tablename", connectionProperties);

// Saving data to a JDBC source
jdbcDF.write()
  .format("jdbc")
  .option("url", "jdbc:postgresql:dbserver")
  .option("dbtable", "schema.tablename")
  .option("user", "username")
  .option("password", "password")
  .save();

jdbcDF2.write()
  .jdbc("jdbc:postgresql:dbserver", "schema.tablename", connectionProperties);

// Specifying create table column data types on write
jdbcDF.write()
  .option("createTableColumnTypes", "name CHAR(64), comments VARCHAR(1024)")
  .jdbc("jdbc:postgresql:dbserver", "schema.tablename", connectionProperties);

3.7 Apache Avro Data Source Guide

• 部署

spark avro模块是外部的，默认情况下不包括在spark submit或spark shell中。
与任何Spark应用程序一样，Spark Submit用于启动应用程序。spark-avro_2.11及其依赖项可以直接添加到spark submit中，使用--packages，例如，

./bin/spark-submit --packages org.apache.spark:spark-avro_2.11:2.4.0 ...

For experimenting on spark-shell, you can also use --packages to add org.apache.spark:spark-avro_2.11 and its dependencies directly,

./bin/spark-shell --packages org.apache.spark:spark-avro_2.11:2.4.0 ...

• Load and Save Functions
  因为spark avro模块是外部的，所以在dataframereader或dataframewriter中没有.avro API。

要以avro格式加载/保存数据，需要将数据源选项格式指定为avro（或org.apache.spark.sql.avro）。

Dataset<Row> usersDF = spark.read().format("avro").load("examples/src/main/resources/users.avro");
usersDF.select("name", "favorite_color").write().format("avro").save("namesAndFavColors.avro");

• to_avro() and from_avro()
  Avro包提供函数to_avro以Avro格式将列编码为二进制，from_avro()将Avro二进制数据解码为列。这两个函数都将一列转换为另一列，输入/输出SQL数据类型可以是复杂类型或原语类型。
  当从Kafka这样的流源读取或写入数据时，使用Avro记录作为列非常有用。每个Kafka键值记录都将添加一些元数据，例如Kafka中的摄入时间戳、Kafka中的偏移量等等。
  如果包含数据的“value”字段位于Avro中，那么可以使用from_avro()来提取数据、丰富数据、清理数据，然后再将其向下推到Kafka或将其写入文件。
  to_avro()可用于将结构体转换为Avro记录。当您在向Kafka写入数据时希望将多个列重新编码为单个列时，此方法尤其有用。
  这两个函数目前只在Scala和Java中可用。

import org.apache.spark.sql.avro.*;

// `from_avro` requires Avro schema in JSON string format.
String jsonFormatSchema = new String(Files.readAllBytes(Paths.get("./examples/src/main/resources/user.avsc")));

Dataset<Row> df = spark
  .readStream()
  .format("kafka")
  .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
  .option("subscribe", "topic1")
  .load();

// 1. Decode the Avro data into a struct;
// 2. Filter by column `favorite_color`;
// 3. Encode the column `name` in Avro format.
Dataset<Row> output = df
  .select(from_avro(col("value"), jsonFormatSchema).as("user"))
  .where("user.favorite_color == \"red\"")
  .select(to_avro(col("user.name")).as("value"));

StreamingQuery query = output
  .writeStream()
  .format("kafka")
  .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
  .option("topic", "topic2")
  .start();

• Data Source Option
  可以使用DataFrameReader或DataFrameWriter上的.option方法设置Avro的数据源选项。

  
  image.png

• 配置
  可以使用SparkSession上的setConf方法或使用SQL运行SET key=value命令来配置Avro。

  
  image.png

• 与Databricks spark-avro的兼容性
  这个Avro数据源模块最初来自Databricks的开源存储库spark-avro，并与之兼容。
  默认情况下，SQL配置为spark.sql.legacy.replaceDatabricksSparkAvro。启用，数据源提供程序com. database .spark。avro映射到这个内置的avro模块。对于使用提供程序属性com. database . Spark创建的Spark表。在catalog元存储中，如果您正在使用这个内置的avro模块，那么映射对于加载这些表是必不可少的。
  注意，在Databricks的spark-avro中，为快捷函数.avro()创建了隐式类AvroDataFrameWriter和AvroDataFrameReader。在这个内置但外部的模块中，这两个隐式类都被删除了。请在DataFrameWriter或DataFrameReader中使用.format("avro")，这样应该足够干净和良好。
  如果您喜欢使用自己构建的spark-avro jar文件，您可以简单地禁用配置spark.sql.legacy.replaceDatabricksSparkAvro。启用，并在部署应用程序时使用选项jar。有关更多细节，请参阅应用程序提交指南中的高级依赖项管理部分。

• Supported types for Avro -> Spark SQL conversion
  目前Spark支持读取Avro记录下的所有原始类型和复杂类型。

  
  image.png
  

  除了上面列出的类型之外，它还支持读取union类型。以下三种类型被认为是基本联合类型:
  1.union(int, long)将映射到LongType。
  2.union(float, double)将映射到DoubleType。
  3.union(something, null)，其中的something是任何受支持的Avro类型。这将映射到与其他类型相同的Spark SQL类型，并将nullable设置为true。所有其他联合类型都被认为是复杂的。它们将被映射到StructType，其中字段名是member0、member1等，根据union的成员。这与在Avro和parquet之间转换时的行为是一致的。

它还支持读取以下Avro逻辑类型:

image.png

• 支持类型的Spark SQL -> Avro转换
  Spark支持将所有Spark SQL类型写入Avro。对于大多数类型，从Spark类型到Avro类型的映射是直接的(例如，IntegerType转换为int);然而，以下列出了一些特殊情况:

  
  image.png
  

  您还可以使用avroSchema选项指定整个输出Avro模式，这样Spark SQL类型就可以转换为其他Avro类型。以下转换在默认情况下不应用，需要用户指定的Avro模式:

  
  image.png

3.8 故障排查

• JDBC驱动程序类必须在客户端会话和所有执行器上对原始类加载程序可见。这是因为Java的DRIVER管理器类做了一个安全检查，导致在打开连接时忽略原始类加载器不可见的所有驱动程序。实现这一点的一个方便方法是修改所有工作节点上的compute_classpath.sh以包括驱动程序jar。
• 一些数据库（如h2）将所有名称转换为大写。您需要使用大写来引用Spark SQL中的那些名称。
• 用户可以在数据源选项中指定特定于供应商的JDBC连接属性来进行特殊处理。例如，spark.read.format（“jdbc”）.option（“url”，oracle jdbc url）.option（“oracle.jdbc.mapDateTimeStamp”，“false”）。Oracle.jdbc.mapDateTimeStamp默认为true，用户通常需要禁用此标志，以避免将Oracle日期解析为时间戳。

4、性能调优

4.1、在内存中缓存数据

Spark SQL可以通过调用Spark .catalog. cachetable(“tableName”)或dataFrame.cache()来使用内存中的柱状格式缓存表。然后Spark SQL将只扫描所需的列，并自动调优压缩，以最小化内存使用和GC压力。可以调用spark.catalog. unachetable(“tableName”)从内存中删除表。
可以使用SparkSession上的setConf方法或使用SQL运行SET key=value命令来配置内存缓存。

image.png

4.2、其他配置选项

还可以使用以下选项来优化查询执行的性能。随着更多优化的自动执行，这些选项可能会在未来的版本中被弃用。

image.png

4.3、用于SQL查询的广播提示

当将每个指定的表与另一个表或视图连接时，BROADCAST提示引导Spark广播它们。当Spark决定连接方法时，广播散列连接(即，最好是BHJ)，即使统计数据高于配置spark.sq . autobroadcastjointhreshold。当指定连接的两边时，Spark广播统计信息较低的一方。注意Spark并不保证总是选择BHJ，因为并非所有情况(例如完全外部连接)都支持BHJ。在选择broadcast嵌套循环连接时，仍然遵循提示。

import static org.apache.spark.sql.functions.broadcast;
broadcast(spark.table("src")).join(spark.table("records"), "key").show();

5、分布式的SQL引擎

Spark SQL还可以使用JDBC/ODBC或命令行接口充当分布式查询引擎。在这种模式下，终端用户或应用程序可以直接与Spark SQL交互来运行SQL查询，而不需要编写任何代码。

5.1、运行Thrift JDBC/ODBC服务器

这里实现的Thrift JDBC/ODBC服务器对应于Hive 1.2.1中的HiveServer2。您可以使用Spark或Hive 1.2.1附带的beeline脚本测试JDBC服务器。
要启动JDBC/ODBC服务器，请在Spark目录中运行以下命令:

./sbin/start-thriftserver.sh

这个脚本接受所有bin/spark-submit命令行选项，加上一个——hiveconf选项来指定Hive属性。您可以运行./sbin/start-thriftserver.sh——帮助获取所有可用选项的完整列表。默认情况下，服务器监听localhost:10000。您可以通过任何环境变量覆盖此行为，即:

export HIVE_SERVER2_THRIFT_PORT=<listening-port>
export HIVE_SERVER2_THRIFT_BIND_HOST=<listening-host>
./sbin/start-thriftserver.sh \
  --master <master-uri> \
  ...

或系统属性:

./sbin/start-thriftserver.sh \
  --hiveconf hive.server2.thrift.port=<listening-port> \
  --hiveconf hive.server2.thrift.bind.host=<listening-host> \
  --master <master-uri>
  ...

现在您可以使用beeline来测试Thrift JDBC/ODBC服务器:

./bin/beeline

#Connect to the JDBC/ODBC server in beeline with:
beeline> !connect jdbc:hive2://localhost:10000

Beeline会询问您的用户名和密码。在非安全模式下，只需在计算机上输入用户名和空白密码。
Hive的配置是通过放置hive-site.xml, core-site.xml and hdfs-site.xml到conf/中。
您还可以使用Hive附带的beeline脚本。
Thrift JDBC server还支持通过HTTP传输发送Thrift RPC消息。使用以下设置将HTTP模式作为系统属性或在hive-site中启用。conf/中的xml文件:

hive.server2.transport.mode - Set this to value: http
hive.server2.thrift.http.port - HTTP port number to listen on; default is 10001
hive.server2.http.endpoint - HTTP endpoint; default is cliservice

要进行测试，请使用beeline在http模式下连接JDBC/ODBC服务器，使用:

beeline> !connect jdbc:hive2://<host>:<port>/<database>?hive.server2.transport.mode=http;hive.server2.thrift.http.path=<http_endpoint>

5.2、运行Spark SQL CLI

Spark SQL CLI是一种方便的工具，可以在本地模式下运行Hive metastore服务并执行命令行输入的查询。请注意，Spark SQL CLI不能与Thrift JDBC服务器通信。
要启动Spark SQL CLI，请在Spark目录中运行以下命令:

./bin/spark-sql

Hive的配置是通过放置hive-site.xml, core-site.xml and hdfs-site.xml到conf/中。

6、参考

6.1、数据类型

Spark SQL和DataFrames支持以下数据类型:

• 数值类型
  ByteType:表示1字节有符号整数。数字的范围是从-128到127。
  ShortType:表示2字节有符号整数。数字的范围从-32768到32767。
  IntegerType:表示4字节有符号整数。数字的范围是从-2147483648到2147483647。
  LongType:表示8字节有符号整数。数字范围从-9223372036854775808到9223372036854775807。
  FloatType:表示4字节的单精度浮点数。
  DoubleType:表示8字节双精度浮点数。
  DecimalType:表示任意精度的带符号小数。内部支持java.math.BigDecimal。BigDecimal由一个任意精度的整数无标度值和一个32位整数标度组成。
• 字符串类型
  StringType:表示字符串值。
• 二进制类型
  BinaryType:表示字节序列值。
• 布尔类型
  BooleanType:表示布尔值。
• Datetime类型
  TimestampType:表示值，包括字段year、month、day、hour、minute和second的值。
  DateType:表示值，其中包含字段year、month、day的值。
• 复杂类型
  ArrayType(elementType, containsNull):表示包含元素序列的值，元素的类型为elementType。containsNull用于指示ArrayType值中的元素是否可以具有空值。
  MapType(keyType, valueType, valueContainsNull):表示由一组键-值对组成的值。键的数据类型由键类型描述，值的数据类型由valueType描述。对于MapType值，键值不允许为空。valueContainsNull用于指示MapType值的值是否可以为空值。
  StructType(fields):用StructFields (fields)序列描述的结构表示值。
  StructField(name, dataType, nullable):表示StructType中的一个字段。字段的名称由名称表示。字段的数据类型由数据类型表示。nullable用于指示此字段的值是否可以为空值。

Spark SQL的所有数据类型都位于org.apache.spark.sql.types的包中。要访问或创建数据类型，请使用org.apache.spark.sql.types.DataTypes中提供的工厂方法。

image.png
image.png

6.1、NaN语义

在处理与标准浮点语义不完全匹配的浮点类型或双类型时，对非数字(NaN)有特殊的处理。具体地说:
1.NaN = NaN返回true。
2.在聚合中，所有NaN值都分组在一起。
3.NaN在连接键中被视为一个正常值。
4.NaN值按升序排列，比任何其他数字值都大，持续时间最长

6.2、算术运算

对数值类型(小数除外)执行的操作不检查溢出。这意味着如果一个操作导致溢出，结果与Java/Scala程序中返回的操作相同。如果两个整数的和大于可表示的最大值，则结果为负数)。