《Spark in Action 第二版》翻译 第三章dataframe的强大作用

LAB 本章的实例可以在GitHub上找到，网址是：https://github.com/jgperrin/net.jgp.books.spark.ch03

dataframe在Spark中的重要作用

dataframe的组织过程

在本节中，你将学习dataframe是如何组织数据的。一个dataframe是一组记录，组织成命名的列。它相当于关系型数据库中的一个表或Java中的ResultSet。图3.2展示了一个dataframe。

不变不是一句脏话

• 你存储数据，每一次修改都会在每个节点上立即完成，就像在关系型数据库中一样。

• 你在各个节点上保持数据同步，只与各个节点共享转换列表。

Spark使用的是第二种解决方案，因为在每个节点上同步一个转换列表比同步所有数据要快。第4章介绍了通过Catalyst 'kæt(ə)lɪst/ 进行优化。Catalyst是Spark处理中负责优化的酷小孩。不变性和转换列表是这个优化引擎的基石。虽然不可变性被Spark出色地用作优化数据处理的基础，但在你开发应用时，你将不必考虑太多。Spark就像任何一个好的操作系统一样，将为你处理资源。

通过实例使用dataframe

一个简单的CSV摄取后的dataframe

LAB 你可以从GitHub上下载代码：https://github.com/jgperrin/net.jgp.books.spark.ch03。这是包net.jgp.books.spark.ch03中的实验室#200.lab200_ingestion_schema_manipulation。

因为分布在多行的记录有点难读，我把记录作为截图添加到图3.9中。

要显示这些数据集（也就是dataframe），你的代码会像下面这样:

到目前为止，这种摄取与第1章摄取简单的书单，以及第2章摄取作者名单的方法类似。摄取的方式总是一样的，第7、8、9章提供了进一步的细节。让我们更深入地了解一下dataframe。你可以通过使用printSchema()将模式打印到标准输出（stdout）。结果如下。

df.printSchema();

System.out.println("We have "+ df.count() + " records.");

• withColumn() 方法-从一个表达式或一个列中创建一个新的列。
• withColumnRenamed() 方法-重命名一个列。
• withColumnRenamed()方法--从列名中获取一个列。有些方法会使用列名作为参数，有些方法需要一个Column对象。
• drop()方法--从数据框架中删除一列。这个方法接受一个列对象的实例或列名。
• lit()函数-创建一个带有值的列；字面意思是一个文本值。
• concat()函数--将一组列中的值进行连接。
  
  

你可能需要为每条记录提供一个唯一的标识符。你可以调用这个列id，并通过连接以下内容来建立它：

1. 国家 

2. 一个下划线(_) 

3. 县 

4. 一个下划线(_) 

5. 数据集中的标识符 

代码是这样的：

    df = df.withColumn("id", concat( df.col("state"), lit("_"), df.col("county"), lit("_"), df.col("datasetId"))); 

最后，你可以显示五条记录并打印模式：

System.out.println("*** Dataframe transformed");
df.show(5);
df.printSchema();

数据是存储在分区中的

df = df.repartition(4);
System.out.println("Partition count after repartition: "+
df.rdd().partitions().length);

探究Schema

*** Schemaas a tree:
        root
|-- OBJECTID: string(nullable = true)
|-- datasetId: string(nullable = true)
|-- name: string(nullable = true)
...

你可以像之前那样使用数据框架的printSchema()方法，或者使用StructType的printTreeString()方法。

你也可以将模式显示为一个简单的字符串。

***** Schema as string: StructField(OBJECTID,StringType,true)StructField(datasetId,StringType,true)StructField(name,StringType,true)... 

要做到这一点，你使用以下代码。

而且你甚至可以将模式显示为JSON结构。

你可以使用以下代码。

更多高级的schema操作，你将在第17章看到。

JSON数据摄取到dataframe后的结构

JSON文件可能比CSV更复杂一些，因为它们的嵌套结构。你要做的实验室与之前类似，但这次餐厅数据的来源是一个JSON文件。本节主要介绍与之前实验室的不同之处，并假设你已经阅读过。使用Spark，你将读取一个JSON文件，其中包含的餐厅数据与3.2.1节中的数据集结构相似。你将对摄入的数据进行转换，以匹配之前数据集的转换结构。你这样做是为了让你可以通过union合并它们。图3.11说明了这个过程的这一部分。

LAB 你可以从GitHub上下载代码：https://github.com/jgperrin/net.jgp.books.spark.ch03。这是包net.jgp.books.spark.ch03中的实验室#220.lab220jsoningestionschemamanipulation。

因为JSON比CSV更难可视化，所以接下来的列表显示了一个只有两家餐厅的数据集节选。JSON绝对更啰嗦，不是吗？我删除了第二条记录的一些字段。

Dataframe包含嵌套的字段和数组。使用show()方法是有用的，但结果不是很好读。对应的schema会给你带来更多的信息。

SpaSparkSession spark = SparkSession.builder().appName("Restaurants in Durham County, NC") .master("local").getOrCreate(); 
Dataset<Row> df = spark.read().format("json").load("data/Restaurants_in_Durham_County_NC.json");
System.out.println("*** Right after ingestion");
df.show(5);df.printSchema(); 

一旦数据在dataframe中，操作数据的API是一样的。你可以开始转换dataframe了。你的目标结构是扁平的，所以映射（如图 3.13 所示）必须包含嵌套字段。

下面是你将产生的内容（图3.14以截图形式显示了dataframe的内容）。

其嵌套结构的schema如下：

要访问结构中的字段，可以在路径中使用点（.）符号。要访问一个数组中的元素，可以使用getItem()方法。下面是实际操作的代码。

就像你创建了所有的字段和列一样，创建id字段的操作和你对CSV文件的操作是一样的:

df = df.withColumn("id", concat(df.col("state"), lit("_"),
df.col("county"), lit("_"), 
df.col("datasetId")));
System.out.println("*** Dataframe transformed");
df.show(5);

最后，看分区也是同样，就像你的CSV文件一样。

这里是代码:

System.out.println("*** Looking at partitions");
Partition[] partitions = df.rdd().partitions();
int partitionCount = partitions.length;
System.out.println("Partition count before repartition: " +  partitionCount);
df = df.repartition(4);
System.out.println("Partition count after repartition: " +  df.rdd().partitions().length);

现在你有了两个dataframe，具有相同的列数据集。下一步是将它们组合起来。

合并两个dataframe

在本节中，您将学习如何在类似于SQL的union来结合两个数据集，以建立一个更大的数据集。这将允许你对更多的数据点进行分析。在上一节中，你摄取了两个数据集，对它们进行了转换，并对它们进行了分析。与关系数据库中的表一样，你可以在它们之间进行很多操作：关联它们、组合它们等等。现在，你要把两个数据集合并起来，这样以后就可以对合并后的数据集进行分析。图3.15说明了这个过程的细节。

LAB 你可以从GitHub上下载代码：https://github.com/jgperrin/net.jgp.books.spark.ch03。这是包net.jgp.books.spark.ch03.lab230dataframeunion中的实验室#230。

你当然可以想象，你可以重用你为摄取和转换所写的大部分代码。然而，要执行union，你必须确保schema是严格相同的。否则，Spark将无法执行union。图 3.16 展示了你要做的映射。

与图3.17对应的模式如下。

• buildWakeRestaurantsDataframe() 构建包含Wake县餐馆的dataframe

• buildDurhamRestaurantsDataframe()建立包含Durham县餐馆的dataframe。

• combineDataframes()通过使用类似SQL的union来合并两个dataframe。现在，不用担心生成的dataframe的内存使用问题。在第4章，你会看到dataframe是自优化的。

这是最简单的部分，对吧？让我们来分析一下这些方法，首先是buildWakeRestaurantsDataframe()，它从CSV文件中读取数据集。这个你应该很熟悉，因为你之前在3.2.1节看到过这个。

现在你已经准备好处理第二个数据集了。

请注意，当你删除一个父列时，所有的嵌套列也会被删除。

现在，你有两个具有相同列数的dataframe。因此，你可以在 combineDataframes() 方法中把它们组合起来。在类似于SQL的组合中，有两种方法可以将两个dataframe组合起来：你可以使用union()或unionByName()方法。union()方法不关心列的名称，只关心它们的顺序。这个方法总是会把第一个dataframe中的第1列和第二个dataframe中的第1列联合起来，然后移动到第2列，再移动到第3列，而不管它们的名字是什么。

Dataframe是一个Dataset<Row>

LAB 你可以从GitHub下载代码，网址是https://github.com/jgperrin/net.jgp.books.spark.ch03。你将从包net.jgp .books.spark.ch03.lab300_dataset中的实验室#300开始。

重复使用您的POJO

Tungsten：疯狂快速的Java存储 性能优化是一个永无止境的故事。Project Tungsten是Apache Spark的一个集成部分，它专注于增强三个关键领域：内存管理和二进制处理，缓存感知计算，以及代码生成。让我们快速看看第一个领域和Java存储对象的方式。在Java中，我喜欢的第一件事（来自C++）是你不必跟踪内存使用和对象生命周期：所有这些都由垃圾收集器（GC）完成。虽然GC在大多数情况下都表现得很好，但当你玩数据集时，它可能很快就会因为创建数百万个对象而不堪重负。在Java（8及以下）中存储一个四字符的字符串，如Java，将需要48个字节；在使用UTF-8/ASCII编码时，存储这个字符串应该只需要4个字节。Java虚拟机(JVM)原生String实现的存储方式不同，它用UTF-16编码对每个字符使用2个字节进行编码，每个String对象还包含一个12字节的头和8字节的哈希码。当你调用Java（或其他任何基于JVM的语言）.length()操作时，JVM仍然会返回4，因为那是以字符为单位的字符串长度，而不是它在内存中的物理表示。查看Java对象布局(JOL)工具，http://openjdk.java.net/projects/code-tools/jol/，了解更多关于物理存储的信息。GC和对象存储本身都不差。然而，在高性能和可预测的工作负载中，本可以取得进步。因此，一个更高效的存储系统诞生了。Tungsten直接管理内存块，压缩数据，并有新的数据容器，使用与操作系统的低级交互，提供16倍到100倍的性能提升。你可以在http://mng.bz/7zyg 阅读更多关于Project Tungsten的信息。

创建一个字符串的dataset

为了理解如何使用dataset而不是dataframe，让我们看看如何创建一个简单的String的dataset。这将通过使用一个我们都熟悉的简单对象--字符串来说明dataset的用法。然后你将能够创建更复杂对象的dataset。你的应用程序将从一个简单的包含字符串的Java数组中创建一个String的dataset，然后显示结果--没有任何花哨的东西。下面是预期的输出。

 Dataset<Row> df = ds.toDF();
 df.show();
 df.printSchema();

输出与本节中的前一个实验室(实验室#300)完全相同，但现在你有了一个dataframe！你可以在这里找到一个dataframe。

来回转换

让我们专注于第一部分：摄取文件，并将dataframe变成书籍的dataset。输出结果如下。

在你将你的dataframe转换为dataset后，字段会被排序。这不是你要求应用程序做的事情；这是一个额外的奖励（或者说 "malus"，取决于日期）。然而，如果你打算在这之后合并数据集，记得使用 unionByName()（而不是 union()），因为字段可能在起飞过程中发生了移动（类似于飞行时东西在头顶的舱室里被移动，这意味着你不知道当你打开盖子时你会得到什么）。操作见以下列表。

• 遍历数据集的每一条记录。
• 在MapFunction类的call()方法中做一些事情。
• 返回一个数据集。

 Dataset<U> map(MapFunction<T, U>, Encoder<U>)

调用时，map()方法将

• 遍历数据框架的每一条记录。

• 调用一个实现MapFunction的类的实例；在你的例子中，就是BookMapper。注意，无论你要处理多少条记录，它都只被实例化一次。

• 返回一个Dataset（你的目标）。

当你实现你的方法时，确保你有正确的签名和实现，因为这可能很棘手。包括签名和所需方法在内的骨架如下：

下面的列表将此骨架应用到您正在构建的BookMapper mapper类中。附录I列出了这些类型的转换的参考资料，包括类签名。

你还需要一个简单的POJO来代表一本书（Book POJO），这是下一个列表。我去掉了大部分的getters和setters，以简化可读性；我很确定你可以在心理上添加缺少的方法。我将所有常见的工件存储在一个x子包中，以增加项目的可读性；在Eclipse中，x代表额外的。

 D Dataset<Row> df2 = bookDs.toDF();

好吧，这并不难，对吧？要将一个dataset转换为dataframe，你只需使用toDF()方法。然而，你仍然有这种奇怪的日期格式，所以我们来纠正一下。第一步是将日期转换为一个具有日期表示的字符串。在这种情况下，你将使用ANSI/ISO格式。YYYY-MM-DD, 如1971-10-05。请记住，Java中的年份是从1900年开始的，所以1971年是71，而2004年是104。同样，月份从0开始，所以10月，也就是一年中的第10个月，是第9个月。使用Java方法来构建日期需要使用一个映射函数，就像你在清单3.3中做的那样。这是通过对数据的迭代来构建数据集或数据框架的方法。你也可以使用UDF，定义在第16章。expr()静态函数将计算一个类似于SQL的表达式并返回一个列。它可以使用字段名。表达式releaseDate.year + 1900将在这个转换过程中被Spark评估，并变成一个包含该值的列。releaseDate.year中的点符号表示数据的路径，你可以在清单3.5的模式中看到。通过实例你将看到更多的静态函数，并将在第13章以及附录G中研究转换。一旦你有了一个作为字符串的日期，你可以使用to_date()静态函数将其转换为作为日期的日期。

你也可以drop() releaseDate列，它的结构很奇怪，它不是很有用。现在你应该可以建立一个包含任何POJO的dataset，并将其转换为一个dataframe。

dataframe的祖先：RDD

• 你不需要模式。
• 你正在开发低级别的转换和动作。
• 你有遗留的代码。

概要

• dataframe是一个不可改变的分布式数据集合，组织成命名的列。基本上，一个dataframe就是一个带有schema的RDD。

• dataframe是Dataset及Row的泛型的一个特例，代码形式为：Dataset<Row>

• 一个dataset除了包含Row形式的数据集，还可以是任何的类型，如Dataset<String>、Dataset<Book>或Dataset<SomePojo>。

• Dataframe可以存储列式信息，就像CSV文件一样，也可以存储嵌套字段和数组，就像JSON文件一样。无论你是在使用CSV文件、JSON文件还是其他格式，dataframe API都是一样的。

• 在JSON文件中，可以使用点号(.)访问嵌套字段。

• dataframe的 API 可以在 http://mng.bz/qXYE 找到；关于如何使用dataframe的细节，请参见参考章节。

• 静态方法的API可以在http://mng.bz/5AQD（和附录G）中找到；关于如何使用静态方法的细节，请参见参考章节。

• 如果你在联合两个dataframe时不关心列名，请使用 union()。

• 如果你在联合两个dataframe时不关心列名，请使用 unionByName()。

• 在Spark中，你可以直接在一个数据集中重用你的POJO。

• 如果你想让一个对象作为数据集的一部分，它必须是可序列化的。

• 数据集的drop()方法可以删除dataframe中的一列。

• 数据集的col()方法根据名称返回数据集的列。

• to_date() 静态函数将字符串中的日期转换为日期。

• expr() 静态函数使用字段名来计算表达式的结果。

• lit() 静态函数返回具有文字值的列。

• 弹性分布式数据集（RDD）是数据元素的不可改变的分布式集合。

• 当性能至关重要时，你应该在RDD上使用dataframe。

• Tungsten存储依赖于dataframe。

• Catalyst 是转换优化器（见第 4 章）。它依靠dataframe来优化操作和转换。

• 跨Spark库的API（graph, SQL, machine learning, 和streaming）正在变得统一在dataframe API下。