SparkSQl 的程序设计

本贴最后更新于 2518 天前,其中的信息可能已经时移俗易

SparkSQL 程序设计流程

  • 1、创建 SparkSession
  • 2、获取数据构建 DF/DS
  • 3、操作数据 DF/DS 获取结果
  • 4、保存数据

1. 创建程序的入口 SparkSession:

import org.apache.spark.sql.SparkSession;

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

2. 构建操作的 DataFrame:

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

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

3. 操作 DataFrame:DSL 操作和 Sql 操作

⬜️ 无类型的 Dataset 算子:

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

⬜️ 创建临时视图 createOrReplaceTempView 首先要将 df 注册为一个 sql 视图

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

⬜️ 全局临时视图 createGlobalTempView 用系统的一个 global___temp 保存数据库,因为普通的视图随会 Session 会话的关闭而消失,如果我们呢需要一个可以保持到整个应用关闭而关闭的视图,那么久可以用“全局临时视图”,使用的时候必须在视图名前 +global___temp,如 global_temp.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,重新创建Session,数据仍然存在
spark.newSession().sql("SELECT * FROM global_temp.people").show();
// +----+-------+
// | age|   name|
// +----+-------+
// |null|Michael|
// |  30|   Andy|
// |  19| Justin|
// +----+-------+

4、构建操作的 Dataset:

Dataset 域 DataFrame 做大的区别在于,DS 是强类型的数据集,编译时的类型检查,所以创建的时候就必须指定数据类型,同时他使用 Endoder 编码器 代替Java序列化和Kryo,去序列化对象,从而实现数据的网络传输,虽然两种编码器都是将对象序列化程字节,如果 Encoder 和标准序列化都能把对象转字节,那么编码器就可以根据代码动态生成,并使用一种特殊数据格式,这种格式下的对象不需要反序列化回来,就能允许 Spark 进行操作,如过滤、排序、哈希等.

Dataset 创建有两个参数:data,和 Encoder,API 如下:Encoder 用于执行数据的类型

626b667982ef44f08b9de7a22495b556-image.png

Encoder:

⬜️ 构建 Object 类型的 Encoder,

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

⬜️ 自带基本数据类型的 Encoder:

  • Encoder integerEncoder = Encoders.INT();
  • Encoder doubleEncoder = Encoders.DOUBLE();
  • BYTE,DATE,FLOAT...
// 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]

⬜️ DataFrame 可以通过指定一个类型来转换成 Dataset.

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

Data:

上面简介了 Encoder 的编码器,这里是 Data 数据来源,有了 data 才能创建 DS,

1、spark 的 read().format('格式').load(path)函数

⬜️ 默认的数据源是: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,非默认数据源要使用 format('类型')指定数据源的类型。

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

2、直接 sql 操作在文件上比先加载再 select 更简便

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

5、RDD->DataFrame->Dataset 转换:

⬜️ 通过反射指定 Schema:如果我们知道 Data 的数据类型是 x,可以使用 createDataFrame(rdd,x.class)

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

// 这里我们知道peopleRDD的schema是People,所以直接指定class
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|
// +------------+

⬜️ 自定义 Schema: 自定义一个 Schema 之后就可以使用:createDataFrame(rdd,schema),当我们不能用一个 JavaBean 作为 Schema 的时候,只能自己定义一个 Schema,分为以下三步:

  • 1、确定字段 Filed 的名称:如'name age gender'
  • 2、构建 StructField:createStructField
  • 3、构建 StructType(即 Schema):DataTypes.createStructType(fields)
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|
// +-------------+

6、数据保存:

保存模式:

Save 操作有一个可选参数 SaveMode,用这个参数可以指定如何处理数据已经存在的情况。很重要的一点是,这些保存模式都没有加锁,所以其操作也不是原子性的。另外,如果使用 Overwrite 模式,实际操作是,先删除数据,再写新数据。

  • SaveMode.ErrorIfExists :(default) "error" (default) (默认模式)从 DataFrame 向数据源保存数据时,如果数据已经存在,则抛异常。

  • SaveMode.Append :"append" 如果数据或表已经存在,则将 DataFrame 的数据追加到已有数据的尾部。

  • SaveMode.Overwrite : "overwrite" 如果数据或表已经存在,则用 DataFrame 数据覆盖之。

  • SaveMode.Ignore : "ignore" 如果数据已经存在,那就放弃保存 DataFrame 数据。这和 SQL 里 CREATE TABLE IF NOT EXISTS 有点类似。

jdbcDF.select("column")
        .sort()
        .as(Encoders.STRING())
        .write()
        .mode(SaveMode.Overwrite) #设置mode
		.json("path")

保存到持久化表

DataFrame 可以使用 saveAsTable 方法将内容持久化到 Hive 的 metastore 表中.默认情况下,saveAsTable 会创建一个”managed table“,也就是说这个表数据的位置是由 metastore 控制的。同样,如果删除表,其数据也会同步删除。

  • Spark

    Spark 是 UC Berkeley AMP lab 所开源的类 Hadoop MapReduce 的通用并行框架。Spark 拥有 Hadoop MapReduce 所具有的优点;但不同于 MapReduce 的是 Job 中间输出结果可以保存在内存中,从而不再需要读写 HDFS,因此 Spark 能更好地适用于数据挖掘与机器学习等需要迭代的 MapReduce 的算法。

    74 引用 • 46 回帖 • 561 关注

相关帖子

欢迎来到这里!

我们正在构建一个小众社区,大家在这里相互信任,以平等 • 自由 • 奔放的价值观进行分享交流。最终,希望大家能够找到与自己志同道合的伙伴,共同成长。

注册 关于
请输入回帖内容 ...