Parquet 文件

peopleDF = spark.read.json("examples/src/main/resources/people.json")
# 数据框可以保存为 Parquet 文件，保持模式信息。
peopleDF.write.parquet("people.parquet")
# 读取上面创建的 Parquet 文件。
# Parquet 文件是自描述的，因此模式得以保留。
# 加载 Parquet 文件的结果也是一个数据框。
parquetFile = spark.read.parquet("people.parquet")
# Parquet 文件也可以用来创建临时视图，然后用于 SQL 语句。
parquetFile.createOrReplaceTempView("parquetFile")
teenagers = spark.sql("SELECT name FROM parquetFile WHERE age >= 13 AND age <= 19")
teenagers.show()
# +------+
# |  name|
# +------+
# |Justin|
# +------+

// 通过导入 spark.implicits._ 自动提供大多数常见类型的编码器
import spark.implicits._
val peopleDF = spark.read.json("examples/src/main/resources/people.json")
// DataFrame 可以作为 Parquet 文件保存，维护模式信息
peopleDF.write.parquet("people.parquet")
// 读取上面创建的 Parquet 文件
// Parquet 文件是自描述的，因此模式得以保留
// 加载 Parquet 文件的结果也是一个 DataFrame
val parquetFileDF = spark.read.parquet("people.parquet")
// Parquet 文件还可以用来创建临时视图，然后在 SQL 语句中使用
parquetFileDF.createOrReplaceTempView("parquetFile")
val namesDF = spark.sql("SELECT name FROM parquetFile WHERE age BETWEEN 13 AND 19")
namesDF.map(attributes => "Name: " + attributes(0)).show()
// +------------+
// |       value|
// +------------+
// |Name: Justin|
// +------------+

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");
// 数据框可以保存为 Parquet 文件，保持架构信息
peopleDF.write().parquet("people.parquet");
// 读取上面创建的 Parquet 文件。
// Parquet 文件是自描述的，因此架构得以保留
// 加载 parquet 文件的结果也是一个 DataFrame
Dataset<Row> parquetFileDF = spark.read().parquet("people.parquet");
// Parquet 文件还可以用于创建临时视图，然后可以在 SQL 语句中使用
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|
// +------------+

df <- read.df("examples/src/main/resources/people.json", "json")
# SparkDataFrame 可以保存为 Parquet 文件，保持模式信息。
write.parquet(df, "people.parquet")
# 读取上述创建的 Parquet 文件。Parquet 文件是自描述的，所以模式得以保留。
# 加载 Parquet 文件的结果也是一个 DataFrame。
parquetFile <- read.parquet("people.parquet")
# Parquet 文件也可以用来创建临时视图，然后可以在 SQL 语句中使用。
createOrReplaceTempView(parquetFile, "parquetFile")
teenagers <- sql("SELECT name FROM parquetFile WHERE age >= 13 AND age <= 19")
head(teenagers)
##     name
## 1 Justin
# 我们还可以在 Spark DataFrames 上运行自定义的 R-UDF。在这里我们为所有名称添加前缀 "Name:"
schema <- structType(structField("name", "string"))
teenNames <- dapply(df, function(p) { cbind(paste("Name:", p$name)) }, schema)
for (teenName in collect(teenNames)$name) {
cat(teenName, "\n")
}
## Name: Michael
## Name: Andy
## Name: Justin

from pyspark.sql import Row
# spark来自前面的例子。
# 创建一个简单的DataFrame，存储到分区目录中
sc = spark.sparkContext
squaresDF = spark.createDataFrame(sc.parallelize(range(1, 6))
.map(lambda i: Row(single=i, double=i ** 2)))
squaresDF.write.parquet("data/test_table/key=1")
# 在一个新的分区目录中创建另一个DataFrame，
# 添加一个新列并删除一个现有列
cubesDF = spark.createDataFrame(sc.parallelize(range(6, 11))
.map(lambda i: Row(single=i, triple=i ** 3)))
cubesDF.write.parquet("data/test_table/key=2")
# 读取分区表
mergedDF = spark.read.option("mergeSchema", "true").parquet("data/test_table")
mergedDF.printSchema()
# 最终的模式由Parquet文件中的所有3列组成
# 以及在分区目录路径中出现的分区列。
# root
#  |-- double: long (nullable = true)
#  |-- single: long (nullable = true)
#  |-- triple: long (nullable = true)
#  |-- key: integer (nullable = true)

// 这用于隐式将 RDD 转换为 DataFrame。
import spark.implicits._
// 创建一个简单的 DataFrame，存储到分区目录中
val squaresDF = spark.sparkContext.makeRDD(1 to 5).map(i => (i, i * i)).toDF("value", "square")
squaresDF.write.parquet("data/test_table/key=1")
// 在一个新的分区目录中创建另一个 DataFrame，
// 添加一个新列并删除一个现有列
val cubesDF = spark.sparkContext.makeRDD(6 to 10).map(i => (i, i * i * i)).toDF("value", "cube")
cubesDF.write.parquet("data/test_table/key=2")
// 读取分区表
val mergedDF = spark.read.option("mergeSchema", "true").parquet("data/test_table")
mergedDF.printSchema()
// 最终的模式由 Parquet 文件中的所有 3 列组成
// 分区列出现在分区目录路径中
// 根
//  |-- value: int (可为空 = true)
//  |-- square: int (可为空 = true)
//  |-- cube: int (可为空 = true)
//  |-- key: int (可为空 = 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;
// 获取器和设置器...
}
public static class Cube implements Serializable {
private int value;
private int cube;
// 获取器和设置器...
}
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);
}
// 创建一个简单的数据框，存储到分区目录中
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);
}
// 在一个新的分区目录中创建另一个数据框，
// 添加一个新列并删除一个现有列
Dataset<Row> cubesDF = spark.createDataFrame(cubes, Cube.class);
cubesDF.write().parquet("data/test_table/key=2");
// 读取分区表
Dataset<Row> mergedDF = spark.read().option("mergeSchema", true).parquet("data/test_table");
mergedDF.printSchema();
// 最终的模式由所有 3 列组成，这些列在 Parquet 文件中一起 
// 以及分区列出现在分区目录路径中
// 根
//  |-- value: int (nullable = true)
//  |-- square: int (nullable = true)
//  |-- cube: int (nullable = true)
//  |-- key: int (nullable = true)

df1 <- createDataFrame(data.frame(single=c(12, 29), double=c(19, 23)))
df2 <- createDataFrame(data.frame(double=c(19, 23), triple=c(23, 18)))
# 创建一个简单的 DataFrame，存储到一个分区目录
write.df(df1, "data/test_table/key=1", "parquet", "overwrite")
# 在新的分区目录中创建另一个 DataFrame，
# 添加一个新列并删除一个现有列
write.df(df2, "data/test_table/key=2", "parquet", "overwrite")
# 读取分区表
df3 <- read.df("data/test_table", "parquet", mergeSchema = "true")
printSchema(df3)
# 最终的模式包括所有 3 列在 Parquet 文件中一起
# 分区列出现在分区目录路径中
## root
##  |-- single: double (nullable = true)
##  |-- double: double (nullable = true)
##  |-- triple: double (nullable = true)
##  |-- key: integer (nullable = true)