Apache Spark：数据框，数据集和RDD之间的区别 - Baeldung


Apache Spark是一个快速的分布式数据处理系统。它执行内存中的数据处理，并使用内存中的缓存和优化的执行，从而实现快速性能。它为流行的编程语言（例如Scala，Python，Java和R）提供了高级API。
在本快速教程中，我们将介绍Spark的三个基本概念：数据帧，数据集和RDD。
 
数据框DataFrame
从Spark 1.3开始，Spark SQL引入了表格形式的数据抽象，称为DataFrame。从那时起，它已成为Spark中最重要的功能之一。当我们要处理结构化和半结构化的分布式数据时，此API很有用。
在第3节中，我们将讨论弹性分布式数据集（RDD）。 DataFrame以比RDD更有效的方式存储数据，这是因为它们使用RDD的不变，内存中，弹性，分布式和并行功能，但它们也将架构应用于数据。DataFrames还可以将SQL代码转换为优化的低级RDD操作。
我们可以通过三种方式创建DataFrame：

  • 转换现有的RDD
  • 运行SQL查询
  • 加载外部数据

Spark团队在2.0版中引入了SparkSession，它统一了所有不同的上下文，从而确保开发人员无需担心创建不同的上下文：
SparkSession session = SparkSession.builder()
  .appName("TouristDataFrameExample")
  .master("local<li>")
  .getOrCreate();
 
DataFrameReader dataFrameReader = session.read();

我们将分析Tourist.csv文件：

Dataset<Row> data = dataFrameReader.option("header", "true")
  .csv("data/Tourist.csv");

由于Spark 2.0 DataFrame成为Row类型的数据集，因此我们可以将DataFrame用作Dataset <Row>的别名。
我们可以选择感兴趣的特定列。我们还可以过滤和分组给定的列：

data.select(col("country"), col("year"), col("value"))
  .show();
 
data.filter(col("country").equalTo("Mexico"))
  .show();
 
data.groupBy(col("country"))
  .count()
  .show();

 

数据集Datasets
数据集是一组强类型的结构化数据。它们提供了熟悉的面向对象编程风格以及类型安全性的好处，因为数据集可以在编译时检查语法并捕获错误。
数据集是DataFrame的扩展，因此我们可以将DataFrame视为数据集的无类型视图。
Spark团队在Spark 1.6中发布了Dataset API，正如他们提到的那样：“ Spark Datasets的目标是提供一个API，使用户可以轻松地表达对象域上的转换，同时还提供Spark SQL执行的性能和鲁棒性优势。发动机”。
首先，我们需要创建一个TouristData类型的类：

public class TouristData {
    private String region;
    private String country;
    private String year;
    private String series;
    private Double value;
    private String footnotes;
    private String source;
    // ... getters and setters
}

要将每个记录映射到指定的类型，我们将需要使用编码器。编码器在Java对象和Spark的内部二进制格式之间进行转换：
// SparkSession initialization and data load
Dataset<Row> responseWithSelectedColumns = data.select(col("region"), 
  col("country"), col("year"), col("series"), col("value").cast("double"), 
  col("footnotes"), col("source"));
 
Dataset<TouristData> typedDataset = responseWithSelectedColumns
  .as(Encoders.bean(TouristData.class));

与DataFrame一样，我们可以按特定的列进行过滤和分组：

typedDataset.filter((FilterFunction) record -> record.getCountry()
  .equals("Norway"))
  .show();
 
typedDataset.groupBy(typedDataset.col("country"))
  .count()
  .show();

我们还可以进行操作，例如按列匹配特定范围进行过滤或计算特定列的总和，以获取其总值：
typedDataset.filter((FilterFunction) record -> record.getYear() != null 
  && (Long.valueOf(record.getYear()) > 2010 
  && Long.valueOf(record.getYear()) < 2017)).show();
 
typedDataset.filter((FilterFunction) record -> record.getValue() != null 
  && record.getSeries()
    .contains("expenditure"))
    .groupBy("country")
    .agg(sum("value"))
    .show();

 
RDD
弹性分布式数据集或RDD是Spark的主要编程抽象。它代表了不可变，有弹性和分布式的元素的集合。
一个RDD封装了一个大型数据集，Spark将自动在整个集群中分布RDD中包含的数据，并并行化我们对其执行的操作。
我们只能通过稳定存储中的数据操作或其他RDD上的操作来创建RDD。
当我们处理大量数据并且数据分布在群集计算机上时，容错能力至关重要。RDD由于Spark内置的故障恢复机制而具有弹性。Spark依赖于以下事实：RDD会记住它们的创建方式，以便我们可以轻松地追溯沿袭来恢复分区。
我们可以对RDD执行两种类型的操作：Transformations和Actions。
  • 转变

我们可以将转换应用于RDD以操纵其数据。执行完此操作后，我们将得到一个全新的RDD，因为RDD是不可变的对象。
我们将检查如何实现Map和Filter，这是两种最常见的转换。
首先，我们需要创建一个JavaSparkContext并从Tourist.csv文件中将数据作为RDD加载：
SparkConf conf = new SparkConf().setAppName("uppercaseCountries")
  .setMaster("local<li>");
JavaSparkContext sc = new JavaSparkContext(conf);
 
JavaRDD<String> tourists = sc.textFile("data/Tourist.csv");

接下来，让我们应用map函数从每个记录中获取国家的名称，并将名称转换为大写。我们可以将此新生成的数据集保存为磁盘上的文本文件：
JavaRDD<String> upperCaseCountries = tourists.map(line -> {
    String[] columns = line.split(COMMA_DELIMITER);
    return columns[1].toUpperCase();
}).distinct();
 
upperCaseCountries.saveAsTextFile("data/output/uppercase.txt");

如果只想选择一个特定国家/地区，则可以对原始游客RDD应用过滤功能：
JavaRDD<String> touristsInMexico = tourists
  .filter(line -> line.split(COMMA_DELIMITER)[1].equals("Mexico"));
 
touristsInMexico.saveAsTextFile("data/output/touristInMexico.txt");

  • Action

在对数据进行一些计算之后，动作将返回最终值或将结果保存到磁盘。
Spark中经常使用的两个动作是Count和Reduce。
让我们在CSV文件中计算国家总数：
// Spark Context initialization and data load
JavaRDD<String> countries = tourists.map(line -> {
    String[] columns = line.split(COMMA_DELIMITER);
    return columns[1];
}).distinct();
 
Long numberOfCountries = countries.count();

现在，我们将按国家/地区计算总支出。我们需要过滤描述中包含支出的记录。
代替使用JavaRDD，我们将使用JavaPairRDD。一对RDD是一种RDD，可以存储键值对。接下来让我们检查一下：

JavaRDD<String> touristsExpenditure = tourists
  .filter(line -> line.split(COMMA_DELIMITER)[3].contains("expenditure"));
 
JavaPairRDD<String, Double> expenditurePairRdd = touristsExpenditure
  .mapToPair(line -> {
      String[] columns = line.split(COMMA_DELIMITER);
      return new Tuple2<>(columns[1], Double.valueOf(columns[6]));
});
 
List<Tuple2<String, Double>> totalByCountry = expenditurePairRdd
  .reduceByKey((x, y) -> x + y)
  .collect();


 综上所述，当我们需要特定于域的API，需要聚合，求和或SQL查询等高级表达式时，应使用DataFrames或Datasets。或者当我们想要在编译时进行类型安全时。
另一方面，当数据是非结构化的并且不需要实现特定的架构时，或者在需要低级的转换和操作时，我们应该使用RDD。
与往常一样，所有代码示例都可以在GitHub上获得。