Hyperspace 使 Apache Spark 用户可以在其数据集（如 CSV、JSON 和 Parquet）上创建索引，并将其用于加速潜在查询和工作负载。
在本文中，我们重点介绍 Hyperspace 的基础知识，展示其简单性，几乎任何人都可以使用它。
免责声明：Hyperspace 在以下两种情况下才有助于加速工作负载或查询：
- 查询包含具有高选择性的谓词上的筛选器。 例如，你可能需要从一百万个候选行中选择 100 个匹配行。
- 查询包含需要大量随机的联接。 例如，你可能需要将 100-GB 数据集和 10-GB 数据集进行联接。
你可能需要仔细监视你的工作负载，并确定索引是否在个别情况下对你有帮助。
本文档也以 Python、C# 和 Scala 的笔记本形式提供。
设置
注意
Azure Synapse Runtime for Apache Spark 3.1（不受支持）和 Azure Synapse Runtime for Apache Spark 3.2（已宣布终止支持）中支持 Hyperspace。 但是，应该注意的是，Azure Synapse Runtime for Apache Spark 3.3 (GA) 不支持 Hyperspace。
首先，启动新的 Spark 会话。 由于此文档是一个教程，仅说明 Hyperspace 可以提供什么，因此你将进行配置更改，使我们能够强调 Hyperspace 在小数据集上所做的工作。
默认情况下，当联接一侧的数据大小较小时（这就是我们在本教程中使用的示例数据的情况），Spark 使用广播联接来优化联接查询。 因此，我们禁用广播联接，以便以后当我们运行联接查询时，Spark 使用排序合并联接。 这主要是为了显示如何大规模使用 Hyperspace 索引来加速联接查询。
运行以下单元的输出将显示对成功创建的 Spark 会话的引用，并输出“-1”作为已修改联接配置的值，这表明广播联接已被成功禁用。
// Start your Spark session
spark
// Disable BroadcastHashJoin, so Spark will use standard SortMergeJoin. Currently, Hyperspace indexes utilize SortMergeJoin to speed up query.
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", -1)
// Verify that BroadcastHashJoin is set correctly
println(spark.conf.get("spark.sql.autoBroadcastJoinThreshold"))
# Start your Spark session.
spark
# Disable BroadcastHashJoin, so Spark will use standard SortMergeJoin. Currently, Hyperspace indexes utilize SortMergeJoin to speed up query.
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", -1)
# Verify that BroadcastHashJoin is set correctly
print(spark.conf.get("spark.sql.autoBroadcastJoinThreshold"))
// Disable BroadcastHashJoin, so Spark will use standard SortMergeJoin. Currently, Hyperspace indexes utilize SortMergeJoin to speed up query.
spark.Conf().Set("spark.sql.autoBroadcastJoinThreshold", -1);
// Verify that BroadcastHashJoin is set correctly.
Console.WriteLine(spark.Conf().Get("spark.sql.autoBroadcastJoinThreshold"));
结果：
res3: org.apache.spark.sql.SparkSession = org.apache.spark.sql.SparkSession@297e957d
-1
数据准备工作
为了准备你的环境，需要创建示例数据记录，并将它们保存为 Parquet 数据文件。 Parquet 只用于说明，但你也可以使用其他格式，如 CSV。 在随后的单元中，你将看到如何在此示例数据集上创建多个 Hyperspace 索引，并在运行查询时让 Spark 使用它们。
示例记录对应于两个数据集：部门和员工。 你应该配置“emp_Location”和“dept_Location”路径，以便在存储帐户上指向你期望的位置，以保存生成的数据文件。
运行以下单元的输出会将我们数据集的内容显示为三元组列表，然后引用为保存首选位置中每个数据集的内容而创建的数据帧。
import org.apache.spark.sql.DataFrame
// Sample department records
val departments = Seq(
(10, "Accounting", "New York"),
(20, "Research", "Dallas"),
(30, "Sales", "Chicago"),
(40, "Operations", "Boston"))
// Sample employee records
val employees = Seq(
(7369, "SMITH", 20),
(7499, "ALLEN", 30),
(7521, "WARD", 30),
(7566, "JONES", 20),
(7698, "BLAKE", 30),
(7782, "CLARK", 10),
(7788, "SCOTT", 20),
(7839, "KING", 10),
(7844, "TURNER", 30),
(7876, "ADAMS", 20),
(7900, "JAMES", 30),
(7934, "MILLER", 10),
(7902, "FORD", 20),
(7654, "MARTIN", 30))
// Save sample data in the Parquet format
import spark.implicits._
val empData: DataFrame = employees.toDF("empId", "empName", "deptId")
val deptData: DataFrame = departments.toDF("deptId", "deptName", "location")
val emp_Location: String = "/<yourpath>/employees.parquet" //TODO ** customize this location path **
val dept_Location: String = "/<yourpath>/departments.parquet" //TODO ** customize this location path **
empData.write.mode("overwrite").parquet(emp_Location)
deptData.write.mode("overwrite").parquet(dept_Location)
from pyspark.sql.types import StructField, StructType, StringType, IntegerType
# Sample department records
departments = [(10, "Accounting", "New York"), (20, "Research", "Dallas"), (30, "Sales", "Chicago"), (40, "Operations", "Boston")]
# Sample employee records
employees = [(7369, "SMITH", 20), (7499, "ALLEN", 30), (7521, "WARD", 30), (7566, "JONES", 20), (7698, "BLAKE", 30)]
# Create a schema for the dataframe
dept_schema = StructType([StructField('deptId', IntegerType(), True), StructField('deptName', StringType(), True), StructField('location', StringType(), True)])
emp_schema = StructType([StructField('empId', IntegerType(), True), StructField('empName', StringType(), True), StructField('deptId', IntegerType(), True)])
departments_df = spark.createDataFrame(departments, dept_schema)
employees_df = spark.createDataFrame(employees, emp_schema)
#TODO ** customize this location path **
emp_Location = "/<yourpath>/employees.parquet"
dept_Location = "/<yourpath>/departments.parquet"
employees_df.write.mode("overwrite").parquet(emp_Location)
departments_df.write.mode("overwrite").parquet(dept_Location)
using Microsoft.Spark.Sql.Types;
// Sample department records
var departments = new List<GenericRow>()
{
new GenericRow(new object[] {10, "Accounting", "New York"}),
new GenericRow(new object[] {20, "Research", "Dallas"}),
new GenericRow(new object[] {30, "Sales", "Chicago"}),
new GenericRow(new object[] {40, "Operations", "Boston"})
};
// Sample employee records
var employees = new List<GenericRow>() {
new GenericRow(new object[] {7369, "SMITH", 20}),
new GenericRow(new object[] {7499, "ALLEN", 30}),
new GenericRow(new object[] {7521, "WARD", 30}),
new GenericRow(new object[] {7566, "JONES", 20}),
new GenericRow(new object[] {7698, "BLAKE", 30}),
new GenericRow(new object[] {7782, "CLARK", 10}),
new GenericRow(new object[] {7788, "SCOTT", 20}),
new GenericRow(new object[] {7839, "KING", 10}),
new GenericRow(new object[] {7844, "TURNER", 30}),
new GenericRow(new object[] {7876, "ADAMS", 20}),
new GenericRow(new object[] {7900, "JAMES", 30}),
new GenericRow(new object[] {7934, "MILLER", 10}),
new GenericRow(new object[] {7902, "FORD", 20}),
new GenericRow(new object[] {7654, "MARTIN", 30})
};
// Save sample data in the Parquet format
var departmentSchema = new StructType(new List<StructField>()
{
new StructField("deptId", new IntegerType()),
new StructField("deptName", new StringType()),
new StructField("location", new StringType())
});
var employeeSchema = new StructType(new List<StructField>()
{
new StructField("empId", new IntegerType()),
new StructField("empName", new StringType()),
new StructField("deptId", new IntegerType())
});
DataFrame empData = spark.CreateDataFrame(employees, employeeSchema);
DataFrame deptData = spark.CreateDataFrame(departments, departmentSchema);
string emp_Location = "/<yourpath>/employees.parquet"; //TODO ** customize this location path **
string dept_Location = "/<yourpath>/departments.parquet"; //TODO ** customize this location path **
empData.Write().Mode("overwrite").Parquet(emp_Location);
deptData.Write().Mode("overwrite").Parquet(dept_Location);
结果：
departments: Seq[(Int, String, String)] = List((10,Accounting,New York), (20,Research,Dallas), (30,Sales,Chicago), (40,Operations,Boston))
employees: Seq[(Int, String, Int)] = List((7369,SMITH,20), (7499,ALLEN,30), (7521,WARD,30), (7566,JONES,20), (7698,BLAKE,30), (7782,CLARK,10), (7788,SCOTT,20), (7839,KING,10), (7844,TURNER,30), (7876,ADAMS,20), (7900,JAMES,30), (7934,MILLER,10), (7902,FORD,20), (7654,MARTIN,30))
empData: org.apache.spark.sql.DataFrame = [empId: int, empName: string ... 1 more field]
deptData: org.apache.spark.sql.DataFrame = [deptId: int, deptName: string ... 1 more field]
emp_Location: String = /your-path/employees.parquet
dept_Location: String = /your-path/departments.parquet
让我们验证所创建的 Parquet 文件的内容，以确保它们包含正确格式的预期记录。 稍后，我们将使用这些数据文件创建 Hyperspace 索引并运行示例查询。
运行下面的单元会生成输出，该输出以表格形式显示员工和部门数据帧中的行。 应该有 14 名员工和 4 个部门，每个部门都与你在上一个单元中创建的三元组之一相匹配。
// emp_Location and dept_Location are the user defined locations above to save parquet files
val empDF: DataFrame = spark.read.parquet(emp_Location)
val deptDF: DataFrame = spark.read.parquet(dept_Location)
// Verify the data is available and correct
empDF.show()
deptDF.show()
# emp_Location and dept_Location are the user-defined locations above to save parquet files
emp_DF = spark.read.parquet(emp_Location)
dept_DF = spark.read.parquet(dept_Location)
# Verify the data is available and correct
emp_DF.show()
dept_DF.show()
// emp_Location and dept_Location are the user-defined locations above to save parquet files
DataFrame empDF = spark.Read().Parquet(emp_Location);
DataFrame deptDF = spark.Read().Parquet(dept_Location);
// Verify the data is available and correct
empDF.Show();
deptDF.Show();
结果：
empDF: org.apache.spark.sql.DataFrame = [empId: int, empName: string ... 1 more field]
deptDF: org.apache.spark.sql.DataFrame = [deptId: int, deptName: string ... 1 more field]
|EmpId|EmpName|DeptId|
|-----|-------|------|
| 7499| ALLEN| 30|
| 7521| WARD| 30|
| 7369| SMITH| 20|
| 7844| TURNER| 30|
| 7876| ADAMS| 20|
| 7900| JAMES| 30|
| 7934| MILLER| 10|
| 7839| KING| 10|
| 7566| JONES| 20|
| 7698| BLAKE| 30|
| 7782| CLARK| 10|
| 7788| SCOTT| 20|
| 7902| FORD| 20|
| 7654| MARTIN| 30|
|DeptId| DeptName|Location|
|------|----------|--------|
| 10|Accounting|New York|
| 40|Operations| Boston|
| 20| Research| Dallas|
| 30| Sales| Chicago|
索引
使用 Hyperspace，你可在从持久数据文件扫描的记录上创建索引。 成功创建后，与索引对应的条目将添加到 Hyperspace 的元数据中。 此元数据后来被 Apache Spark 的优化器（与扩展一起）用于在查询处理期间查找和使用正确的索引。
创建索引后，可以执行多个操作：
- 如果基础数据发生更改则刷新。 可以刷新现有索引来捕获更改。
- 如果索引不是必需的则删除。 你可以执行软删除，即不是以物理方式删除索引，而是将其标记为“已删除”，以便它不再用于你的工作负载。
- 如果不再需要索引则清空。 你可以清空索引，这会强制从 Hyperspace 的元数据中完全物理删除索引内容和关联的元数据。
如果基础数据发生更改则刷新，你可以刷新现有索引以捕获该索引。 如果索引不是必需的则删除，你可以执行软删除，即不是以物理方式删除索引，而是将其标记为“已删除”，以便它不再用于你的工作负载。
以下部分显示了如何在 Hyperspace 中执行此类索引管理操作。
首先，你需要导入所需的库并创建 Hyperspace 实例。 稍后，你将使用此实例调用不同的 Hyperspace API 来创建样示例数据上的索引并修改这些索引。
运行以下单元的输出将显示对 Hyperspace 已创建实例的引用。
// Create an instance of Hyperspace
import com.microsoft.hyperspace._
val hyperspace: Hyperspace = Hyperspace()
from hyperspace import *
# Create an instance of Hyperspace
hyperspace = Hyperspace(spark)
// Create an instance of Hyperspace
using Microsoft.Spark.Extensions.Hyperspace;
Hyperspace hyperspace = new Hyperspace(spark);
结果：
hyperspace: com.microsoft.hyperspace.Hyperspace = com.microsoft.hyperspace.Hyperspace@1432f740
创建索引
若要创建 Hyperspace 索引，你需要提供两条信息：
- 一个 Spark 数据帧，用于引用要编制索引的数据。
- 一个索引配置对象 IndexConfig，用于指定索引名称以及索引的索引列和包含列。
首先，在我们的样本数据上创建三个 Hyperspace 索引：部门数据集上的名为“deptIndex1”和“deptIndex2”的两个索引，员工数据集上的名为“empIndex”的一个索引。 对于每个索引，需要一个相应的 IndexConfig 来捕获名称以及索引列和包含列的列列表。 运行以下单元将创建这些 IndexConfig，其输出会列出它们。
注意
索引列是显示在筛选器或联接条件中的列。 包含列是显示在你的选择/项目中的列。
例如，在以下查询中：
SELECT X
FROM T
WHERE Y = 2
Y 可以是索引列，X 可以是包含列。
// Create index configurations
import com.microsoft.hyperspace.index.IndexConfig
val empIndexConfig: IndexConfig = IndexConfig("empIndex", Seq("deptId"), Seq("empName"))
val deptIndexConfig1: IndexConfig = IndexConfig("deptIndex1", Seq("deptId"), Seq("deptName"))
val deptIndexConfig2: IndexConfig = IndexConfig("deptIndex2", Seq("location"), Seq("deptName"))
# Create index configurations
emp_IndexConfig = IndexConfig("empIndex1", ["deptId"], ["empName"])
dept_IndexConfig1 = IndexConfig("deptIndex1", ["deptId"], ["deptName"])
dept_IndexConfig2 = IndexConfig("deptIndex2", ["location"], ["deptName"])
using Microsoft.Spark.Extensions.Hyperspace.Index;
var empIndexConfig = new IndexConfig("empIndex", new string[] {"deptId"}, new string[] {"empName"});
var deptIndexConfig1 = new IndexConfig("deptIndex1", new string[] {"deptId"}, new string[] {"deptName"});
var deptIndexConfig2 = new IndexConfig("deptIndex2", new string[] {"location"}, new string[] {"deptName"});
结果：
empIndexConfig: com.microsoft.hyperspace.index.IndexConfig = [indexName: empIndex; indexedColumns: deptid; includedColumns: empname]
deptIndexConfig1: com.microsoft.hyperspace.index.IndexConfig = [indexName: deptIndex1; indexedColumns: deptid; includedColumns: deptname]
deptIndexConfig2: com.microsoft.hyperspace.index.IndexConfig = [indexName: deptIndex2; indexedColumns: location; includedColumns: deptname]
现在，你可以使用索引配置创建三个索引。 为此，请在我们的 Hyperspace 实例中调用“createIndex”命令。 此命令需要索引配置和包含行的数据帧进行编制索引。 运行以下单元将创建三个索引。
// Create indexes from configurations
import com.microsoft.hyperspace.index.Index
hyperspace.createIndex(empDF, empIndexConfig)
hyperspace.createIndex(deptDF, deptIndexConfig1)
hyperspace.createIndex(deptDF, deptIndexConfig2)
# Create indexes from configurations
hyperspace.createIndex(emp_DF, emp_IndexConfig)
hyperspace.createIndex(dept_DF, dept_IndexConfig1)
hyperspace.createIndex(dept_DF, dept_IndexConfig2)
// Create indexes from configurations
hyperspace.CreateIndex(empDF, empIndexConfig);
hyperspace.CreateIndex(deptDF, deptIndexConfig1);
hyperspace.CreateIndex(deptDF, deptIndexConfig2);
列出索引
下面的代码显示了如何在 Hyperspace 实例中列出所有可用的索引。 它使用“索引”API 将现有索引信息作为 Spark DataFrame 返回，以便你可以执行更多操作。
例如，你可以调用此数据帧上的有效操作来检查其内容或进一步分析它（例如，筛选特定索引或根据某些所需属性对它们进行分组）。
以下单元使用数据帧的“显示”操作以表格形式完全输出行并显示我们索引的详细信息。 对于每个索引，你可以在元数据中查看 Hyperspace 存储的所有信息。 你将立即注意到以下内容：
- config.indexName、config.indexedColumns、config.includedColumns 和 status.status 是用户通常引用的字段。
- dfSignature 由 Hyperspace 自动生成，并且对于每个索引都是唯一的。 Hyperspace 在内部使用此签名来维护索引并在查询时间进行利用。
在以下输出中，所有三个索引都应具有“可用”状态及其名称、索引列，并且包含列应与我们在上面的索引配置中定义的内容匹配。
hyperspace.indexes.show
hyperspace.indexes().show()
hyperspace.Indexes().Show();
结果：
|Config.IndexName|Config.IndexedColumns|Config.IncludedColumns| SchemaString| SignatureProvider| DfSignature| SerializedPlan|NumBuckets| DirPath|Status.Value|Stats.IndexSize|
|----------------|---------------------|----------------------|--------------------|--------------------|--------------------|--------------------|----------|--------------------|------------|---------------|
| deptIndex1| [deptId]| [deptName]|`deptId` INT,`dep...|com.microsoft.cha...|0effc1610ae2e7c49...|Relation[deptId#3...| 200|abfss://datasets@...| ACTIVE| 0|
| deptIndex2| [location]| [deptName]|`location` STRING...|com.microsoft.cha...|0effc1610ae2e7c49...|Relation[deptId#3...| 200|abfss://datasets@...| ACTIVE| 0|
| empIndex| [deptId]| [empName]|`deptId` INT,`emp...|com.microsoft.cha...|30768c6c9b2533004...|Relation[empId#32...| 200|abfss://datasets@...| ACTIVE| 0|
删除索引
你可以使用“deleteIndex”API 并提供索引名称来删除现有索引。 索引删除会进行软删除：它主要将索引在 Hyperspace 元数据中的状态从“可用”更新为“已删除”。 这将排除任何未来查询优化中删除的索引，Hyperspace 不再为任何查询选择该索引。
但是，已删除索引的索引文件仍然可用（因为它是软删除），这样如果用户要求，可以恢复索引。
以下单元将删除名为“deptIndex2”的索引，并在此之后列出 Hyperspace 元数据。 输出应类似于“列出索引”的上述单元格，但“deptIndex2”除外，后者现在应将其状态更改为“已删除”。
hyperspace.deleteIndex("deptIndex2")
hyperspace.indexes.show
hyperspace.deleteIndex("deptIndex2")
hyperspace.indexes().show()
hyperspace.DeleteIndex("deptIndex2");
hyperspace.Indexes().Show();
结果：
|Config.IndexName|Config.IndexedColumns|Config.IncludedColumns| SchemaString| SignatureProvider| DfSignature| SerializedPlan|NumBuckets| DirPath|Status.Value|Stats.IndexSize|
|----------------|---------------------|----------------------|--------------------|--------------------|--------------------|--------------------|----------|--------------------|------------|---------------|
| deptIndex1| [deptId]| [deptName]|`deptId` INT,`dep...|com.microsoft.cha...|0effc1610ae2e7c49...|Relation[deptId#3...| 200|abfss://datasets@...| ACTIVE| 0|
| deptIndex2| [location]| [deptName]|`location` STRING...|com.microsoft.cha...|0effc1610ae2e7c49...|Relation[deptId#3...| 200|abfss://datasets@...| DELETED| 0|
| empIndex| [deptId]| [empName]|`deptId` INT,`emp...|com.microsoft.cha...|30768c6c9b2533004...|Relation[empId#32...| 200|abfss://datasets@...| ACTIVE| 0|
还原索引
可以使用“restoreIndex”API 来还原已删除的索引。 这将使最新版本的索引恢复为“可用”状态，并使其再次用于查询。 以下单元将显示“restoreIndex”使用情况的示例。 删除“deptIndex1”并将其还原。 输出显示“deptIndex1”在调用“deleteIndex”命令后首次进入“已删除”状态，并在调用“restoreIndex”后返回到“可用”状态。
hyperspace.deleteIndex("deptIndex1")
hyperspace.indexes.show
hyperspace.restoreIndex("deptIndex1")
hyperspace.indexes.show
hyperspace.deleteIndex("deptIndex1")
hyperspace.indexes().show()
hyperspace.restoreIndex("deptIndex1")
hyperspace.indexes().show()
hyperspace.DeleteIndex("deptIndex1");
hyperspace.Indexes().Show();
hyperspace.RestoreIndex("deptIndex1");
hyperspace.Indexes().Show();
结果：
|Config.IndexName|Config.IndexedColumns|Config.IncludedColumns| SchemaString| SignatureProvider| DfSignature| SerializedPlan|NumBuckets| DirPath|Status.Value|Stats.indexSize|
|----------------|---------------------|----------------------|--------------------|--------------------|--------------------|--------------------|----------|--------------------|------------|---------------|
| deptIndex1| [deptId]| [deptName]|`deptId` INT,`dep...|com.microsoft.cha...|0effc1610ae2e7c49...|Relation[deptId#3...| 200|abfss://datasets@...| DELETED| 0|
| deptIndex2| [location]| [deptName]|`location` STRING...|com.microsoft.cha...|0effc1610ae2e7c49...|Relation[deptId#3...| 200|abfss://datasets@...| DELETED| 0|
| empIndex| [deptId]| [empName]|`deptId` INT,`emp...|com.microsoft.cha...|30768c6c9b2533004...|Relation[empId#32...| 200|abfss://datasets@...| ACTIVE| 0|
|Config.IndexName|Config.IndexedColumns|Config.IncludedColumns| SchemaString| SignatureProvider| DfSignature| SerializedPlan|NumBuckets| DirPath|Status.value|Stats.IndexSize|
|----------------|---------------------|----------------------|--------------------|--------------------|--------------------|--------------------|----------|--------------------|------------|---------------|
| deptIndex1| [deptId]| [deptName]|`deptId` INT,`dep...|com.microsoft.cha...|0effc1610ae2e7c49...|Relation[deptId#3...| 200|abfss://datasets@...| ACTIVE| 0|
| deptIndex2| [location]| [deptName]|`location` STRING...|com.microsoft.cha...|0effc1610ae2e7c49...|Relation[deptId#3...| 200|abfss://datasets@...| DELETED| 0|
| empIndex| [deptId]| [empName]|`deptId` INT,`emp...|com.microsoft.cha...|30768c6c9b2533004...|Relation[empId#32...| 200|abfss://datasets@...| ACTIVE| 0|
清空索引
你可以使用“vacuumIndex”命令执行硬删除，即完全删除已删除索引的文件和元数据条目。 此操作不可逆。 它以物理方式删除所有索引文件，这就是为什么它是硬删除。
以下单元将清空“deptIndex2”索引，并在清空后显示 Hyperspace 元数据。 你应看到两个索引“deptIndex1”和“empIndex”的元数据条目，这两个索引都具有“可用”状态，并且没有“deptIndex2”的条目。
hyperspace.vacuumIndex("deptIndex2")
hyperspace.indexes.show
hyperspace.vacuumIndex("deptIndex2")
hyperspace.indexes().show()
hyperspace.VacuumIndex("deptIndex2");
hyperspace.Indexes().Show();
结果：
|Config.IndexName|Config.IndexedColumns|Config.IncludedColumns| SchemaString| SignatureProvider| DfSignature| SerializedPlan|NumBuckets| DirPath|Status.Value|Stats.IndexSize|
|----------------|---------------------|----------------------|--------------------|--------------------|--------------------|--------------------|----------|--------------------|------------|---------------|
| deptIndex1| [deptId]| [deptName]|`deptId` INT,`dep...|com.microsoft.cha...|0effc1610ae2e7c49...|Relation[deptId#3...| 200|abfss://datasets@...| ACTIVE| 0|
| empIndex| [deptId]| [empName]|`deptId` INT,`emp...|com.microsoft.cha...|30768c6c9b2533004...|Relation[empId#32...| 200|abfss://datasets@...| ACTIVE| 0|
启用或禁用 Hyperspace
Hyperspace 提供通过 Spark 启用或禁用索引使用的 API。
- 通过使用 enableHyperspace 命令，Spark 优化器可以看到 Hyperspace 优化规则，并利用现有的 Hyperspace 索引来优化用户查询。
- 通过使用 disableHyperspace 命令，在查询优化过程中 Hyperspace 规则将不再适用。 禁用 Hyperspace 对创建的索引没有影响，因为它们保持不变。
以下单元显示如何使用这些命令启用或禁用 Hyperspace。 输出显示对已更新配置的现有 Spark 会话的引用。
// Enable Hyperspace
spark.enableHyperspace
// Disable Hyperspace
spark.disableHyperspace
# Enable Hyperspace
Hyperspace.enable(spark)
# Disable Hyperspace
Hyperspace.disable(spark)
// Enable Hyperspace
spark.EnableHyperspace();
// Disable Hyperspace
spark.DisableHyperspace();
结果：
res48: org.apache.spark.sql.Spark™Session = org.apache.spark.sql.SparkSession@39fe1ddb
res51: org.apache.spark.sql.Spark™Session = org.apache.spark.sql.SparkSession@39fe1ddb
索引使用情况
若要使 Spark 在查询处理过程中使用 Hyperspace 索引，你需要确保已启用 Hyperspace。
以下单元启用 Hyperspace，并创建两个包含示例数据记录的数据帧，用于运行示例查询。 对于每个数据帧，都会输出几个示例行。
// Enable Hyperspace
spark.enableHyperspace
val empDFrame: DataFrame = spark.read.parquet(emp_Location)
val deptDFrame: DataFrame = spark.read.parquet(dept_Location)
empDFrame.show(5)
deptDFrame.show(5)
# Enable Hyperspace
Hyperspace.enable(spark)
emp_DF = spark.read.parquet(emp_Location)
dept_DF = spark.read.parquet(dept_Location)
emp_DF.show(5)
dept_DF.show(5)
// Enable Hyperspace
spark.EnableHyperspace();
DataFrame empDFrame = spark.Read().Parquet(emp_Location);
DataFrame deptDFrame = spark.Read().Parquet(dept_Location);
empDFrame.Show(5);
deptDFrame.Show(5);
结果：
res53: org.apache.spark.sql.Spark™Session = org.apache.spark.sql.Spark™Session@39fe1ddb
empDFrame: org.apache.spark.sql.DataFrame = [empId: int, empName: string ... 1 more field]
deptDFrame: org.apache.spark.sql.DataFrame = [deptId: int, deptName: string ... 1 more field]
|empId|empName|deptId|
|-----|-------|------|
| 7499| ALLEN| 30|
| 7521| WARD| 30|
| 7369| SMITH| 20|
| 7844| TURNER| 30|
| 7876| ADAMS| 20|
这仅显示前 5 行
|deptId| deptName|location|
|------|----------|--------|
| 10|Accounting|New York|
| 40|Operations| Boston|
| 20| Research| Dallas|
| 30| Sales| Chicago|
索引类型
目前，Hyperspace 包含利用两组查询的索引的规则：
- 带有查找或范围选择筛选谓词的选择查询。
- 带有相等联接谓词的查询的联接查询（即同等联接）。
用于加速筛选器的索引
第一个示例查询查找部门记录，如以下单元所示。 在 SQL 中，该查询如以下示例所示：
SELECT deptName
FROM departments
WHERE deptId = 20
运行以下单元的输出显示：
- 查询结果，它是一个部门名称。
- Spark 用于运行查询的查询计划。
在查询计划中，计划底部的 FileScan 运算符显示从中读取记录的数据源。 此文件的位置表示最新版本的“deptIndex1”索引的路径。 此信息表明，根据查询和使用 Hyperspace 优化规则，Spark 决定在运行时利用合适的索引。
// Filter with equality predicate
val eqFilter: DataFrame = deptDFrame.filter("deptId = 20").select("deptName")
eqFilter.show()
eqFilter.explain(true)
# Filter with equality predicate
eqFilter = dept_DF.filter("""deptId = 20""").select("""deptName""")
eqFilter.show()
eqFilter.explain(True)
DataFrame eqFilter = deptDFrame.Filter("deptId = 20").Select("deptName");
eqFilter.Show();
eqFilter.Explain(true);
结果：
eqFilter: org.apache.spark.sql.DataFrame = [deptName: string]
|DeptName|
|--------|
|Research|
== Parsed Logical Plan ==
'Project [unresolvedalias('deptName, None)]
+- Filter (deptId#533 = 20)
+- Relation[deptId#533,deptName#534,location#535] parquet
== Analyzed Logical Plan ==
deptName: string
Project [deptName#534]
+- Filter (deptId#533 = 20)
+- Relation[deptId#533,deptName#534,location#535] parquet
== Optimized Logical Plan ==
Project [deptName#534]
+- Filter (isnotnull(deptId#533) && (deptId#533 = 20))
+- Relation[deptId#533,deptName#534,location#535] parquet
== Physical Plan ==
*(1) Project [deptName#534]
+- *(1) Filter (isnotnull(deptId#533) && (deptId#533 = 20))
+- *(1) FileScan parquet [deptId#533,deptName#534] Batched: true, Format: Parquet, Location: InMemoryFileIndex[abfss://datasets@hyperspacebenchmark.dfs.core.chinacloudapi.cn/hyperspaceon..., PartitionFilters: [], PushedFilters: [IsNotNull(deptId), EqualTo(deptId,20)], ReadSchema: struct<deptId:int,deptName:string>
第二个例子是部门记录上的范围选择查询。 在 SQL 中，该查询如以下示例所示：
SELECT deptName
FROM departments
WHERE deptId > 20
与第一个示例类似，以下单元的输出显示查询结果（两个部门的名称）和查询计划。 FileScan 运算符中的数据文件的位置显示，“deptIndex1”用于运行查询。
// Filter with range selection predicate
val rangeFilter: DataFrame = deptDFrame.filter("deptId > 20").select("deptName")
rangeFilter.show()
rangeFilter.explain(true)
# Filter with range selection predicate
rangeFilter = dept_DF.filter("""deptId > 20""").select("deptName")
rangeFilter.show()
rangeFilter.explain(True)
// Filter with range selection predicate
DataFrame rangeFilter = deptDFrame.Filter("deptId > 20").Select("deptName");
rangeFilter.Show();
rangeFilter.Explain(true);
结果：
rangeFilter: org.apache.spark.sql.DataFrame = [deptName: string]
| DeptName|
|----------|
|Operations|
| Sales|
== Parsed Logical Plan ==
'Project [unresolvedalias('deptName, None)]
+- Filter (deptId#533 > 20)
+- Relation[deptId#533,deptName#534,location#535] parquet
== Analyzed Logical Plan ==
deptName: string
Project [deptName#534]
+- Filter (deptId#533 > 20)
+- Relation[deptId#533,deptName#534,location#535] parquet
== Optimized Logical Plan ==
Project [deptName#534]
+- Filter (isnotnull(deptId#533) && (deptId#533 > 20))
+- Relation[deptId#533,deptName#534,location#535] parquet
== Physical Plan ==
*(1) Project [deptName#534]
+- *(1) Filter (isnotnull(deptId#533) && (deptId#533 > 20))
+- *(1) FileScan parquet [deptId#533,deptName#534] Batched: true, Format: Parquet, Location: InMemoryFileIndex[abfss://datasets@hyperspacebenchmark.dfs.core.chinacloudapi.cn/hyperspaceon..., PartitionFilters: [], PushedFilters: [IsNotNull(deptId), GreaterThan(deptId,20)], ReadSchema: struct<deptId:int,deptName:string>
第三个例子是部门 ID 上的查询联接部门和员工记录。 等效的 SQL 语句如下所示：
SELECT employees.deptId, empName, departments.deptId, deptName
FROM employees, departments
WHERE employees.deptId = departments.deptId
运行以下单元的输出显示查询结果，其为 14 名员工的姓名和每个员工所在的部门的名称。 查询计划也包含在输出中。 请注意两个 FileScan 运算符的文件位置如何显示 Spark 使用“empIndex”和“deptIndex1”索引来运行查询。
// Join
val eqJoin: DataFrame =
empDFrame.
join(deptDFrame, empDFrame("deptId") === deptDFrame("deptId")).
select(empDFrame("empName"), deptDFrame("deptName"))
eqJoin.show()
eqJoin.explain(true)
# Join
eqJoin = emp_DF.join(dept_DF, emp_DF.deptId == dept_DF.deptId).select(emp_DF.empName, dept_DF.deptName)
eqJoin.show()
eqJoin.explain(True)
// Join
DataFrame eqJoin =
empDFrame
.Join(deptDFrame, empDFrame.Col("deptId") == deptDFrame.Col("deptId"))
.Select(empDFrame.Col("empName"), deptDFrame.Col("deptName"));
eqJoin.Show();
eqJoin.Explain(true);
结果：
eqJoin: org.apache.spark.sql.DataFrame = [empName: string, deptName: string]
|empName| deptName|
|-------|----------|
| SMITH| Research|
| JONES| Research|
| FORD| Research|
| ADAMS| Research|
| SCOTT| Research|
| KING|Accounting|
| CLARK|Accounting|
| MILLER|Accounting|
| JAMES| Sales|
| BLAKE| Sales|
| MARTIN| Sales|
| ALLEN| Sales|
| WARD| Sales|
| TURNER| Sales|
== Parsed Logical Plan ==
Project [empName#528, deptName#534]
+- Join Inner, (deptId#529 = deptId#533)
:- Relation[empId#527,empName#528,deptId#529] parquet
+- Relation[deptId#533,deptName#534,location#535] parquet
== Analyzed Logical Plan ==
empName: string, deptName: string
Project [empName#528, deptName#534]
+- Join Inner, (deptId#529 = deptId#533)
:- Relation[empId#527,empName#528,deptId#529] parquet
+- Relation[deptId#533,deptName#534,location#535] parquet
== Optimized Logical Plan ==
Project [empName#528, deptName#534]
+- Join Inner, (deptId#529 = deptId#533)
:- Project [empName#528, deptId#529]
: +- Filter isnotnull(deptId#529)
: +- Relation[empName#528,deptId#529] parquet
+- Project [deptId#533, deptName#534]
+- Filter isnotnull(deptId#533)
+- Relation[deptId#533,deptName#534] parquet
== Physical Plan ==
*(3) Project [empName#528, deptName#534]
+- *(3) SortMergeJoin [deptId#529], [deptId#533], Inner
:- *(1) Project [empName#528, deptId#529]
: +- *(1) Filter isnotnull(deptId#529)
: +- *(1) FileScan parquet [deptId#529,empName#528] Batched: true, Format: Parquet, Location: InMemoryFileIndex[abfss://datasets@hyperspacebenchmark.dfs.core.chinacloudapi.cn/hyperspaceon..., PartitionFilters: [], PushedFilters: [IsNotNull(deptId)], ReadSchema: struct<deptId:int,empName:string>, SelectedBucketsCount: 200 out of 200
+- *(2) Project [deptId#533, deptName#534]
+- *(2) Filter isnotnull(deptId#533)
+- *(2) FileScan parquet [deptId#533,deptName#534] Batched: true, Format: Parquet, Location: InMemoryFileIndex[abfss://datasets@hyperspacebenchmark.dfs.core.chinacloudapi.cn/hyperspaceon..., PartitionFilters: [], PushedFilters: [IsNotNull(deptId)], ReadSchema: struct<deptId:int,deptName:string>, SelectedBucketsCount: 200 out of 200
支持 SQL 语义
索引的使用对于你是否使用数据帧 API 或 Spark SQL 是透明的。 以下示例以 SQL 形式显示与以前相同的联接示例，从而显示索引的使用情况（如果适用）。
empDFrame.createOrReplaceTempView("EMP")
deptDFrame.createOrReplaceTempView("DEPT")
val joinQuery = spark.sql("SELECT EMP.empName, DEPT.deptName FROM EMP, DEPT WHERE EMP.deptId = DEPT.deptId")
joinQuery.show()
joinQuery.explain(true)
from pyspark.sql import SparkSession
emp_DF.createOrReplaceTempView("EMP")
dept_DF.createOrReplaceTempView("DEPT")
joinQuery = spark.sql("SELECT EMP.empName, DEPT.deptName FROM EMP, DEPT WHERE EMP.deptId = DEPT.deptId")
joinQuery.show()
joinQuery.explain(True)
empDFrame.CreateOrReplaceTempView("EMP");
deptDFrame.CreateOrReplaceTempView("DEPT");
var joinQuery = spark.Sql("SELECT EMP.empName, DEPT.deptName FROM EMP, DEPT WHERE EMP.deptId = DEPT.deptId");
joinQuery.Show();
joinQuery.Explain(true);
结果：
joinQuery: org.apache.spark.sql.DataFrame = [empName: string, deptName: string]
|empName| deptName|
|-------|----------|
| SMITH| Research|
| JONES| Research|
| FORD| Research|
| ADAMS| Research|
| SCOTT| Research|
| KING|Accounting|
| CLARK|Accounting|
| MILLER|Accounting|
| JAMES| Sales|
| BLAKE| Sales|
| MARTIN| Sales|
| ALLEN| Sales|
| WARD| Sales|
| TURNER| Sales|
== Parsed Logical Plan ==
'Project ['EMP.empName, 'DEPT.deptName]
+- 'Filter ('EMP.deptId = 'DEPT.deptId)
+- 'Join Inner
:- 'UnresolvedRelation `EMP`
+- 'UnresolvedRelation `DEPT`
== Analyzed Logical Plan ==
empName: string, deptName: string
Project [empName#528, deptName#534]
+- Filter (deptId#529 = deptId#533)
+- Join Inner
:- SubqueryAlias `emp`
: +- Relation[empId#527,empName#528,deptId#529] parquet
+- SubqueryAlias `dept`
+- Relation[deptId#533,deptName#534,location#535] parquet
== Optimized Logical Plan ==
Project [empName#528, deptName#534]
+- Join Inner, (deptId#529 = deptId#533)
:- Project [empName#528, deptId#529]
: +- Filter isnotnull(deptId#529)
: +- Relation[empId#527,empName#528,deptId#529] parquet
+- Project [deptId#533, deptName#534]
+- Filter isnotnull(deptId#533)
+- Relation[deptId#533,deptName#534,location#535] parquet
== Physical Plan ==
*(5) Project [empName#528, deptName#534]
+- *(5) SortMergeJoin [deptId#529], [deptId#533], Inner
:- *(2) Sort [deptId#529 ASC NULLS FIRST], false, 0
: +- Exchange hashpartitioning(deptId#529, 200)
: +- *(1) Project [empName#528, deptId#529]
: +- *(1) Filter isnotnull(deptId#529)
: +- *(1) FileScan parquet [deptId#529,empName#528] Batched: true, Format: Parquet, Location: InMemoryFileIndex[abfss://datasets@hyperspacebenchmark.dfs.core.chinacloudapi.cn/hyperspaceon..., PartitionFilters: [], PushedFilters: [IsNotNull(deptId)], ReadSchema: struct<deptId:int,empName:string>
+- *(4) Sort [deptId#533 ASC NULLS FIRST], false, 0
+- Exchange hashpartitioning(deptId#533, 200)
+- *(3) Project [deptId#533, deptName#534]
+- *(3) Filter isnotnull(deptId#533)
+- *(3) FileScan parquet [deptId#533,deptName#534] Batched: true, Format: Parquet, Location: InMemoryFileIndex[abfss://datasets@hyperspacebenchmark.dfs.core.chinacloudapi.cn/your-path/departments.parquet], PartitionFilters: [], PushedFilters: [IsNotNull(deptId)], ReadSchema: struct<deptId:int,deptName:string>
说明 API
索引非常好，但你怎么知道它们是否正在被使用？ 通过 Hyperspace，用户可以在运行查询之前比较其原始计划与更新的依赖索引的计划。 你可以选择使用 HTML、纯文本或控制台模式来显示命令输出。
以下单元显示了使用 HTML 的示例。 突出显示的部分表示原始计划和更新计划以及正在使用的索引之间的差异。
spark.conf.set("spark.hyperspace.explain.displayMode", "html")
hyperspace.explain(eqJoin)(displayHTML(_))
eqJoin = emp_DF.join(dept_DF, emp_DF.deptId == dept_DF.deptId).select(emp_DF.empName, dept_DF.deptName)
spark.conf.set("spark.hyperspace.explain.displayMode", "html")
hyperspace.explain(eqJoin, True, displayHTML)
spark.Conf().Set("spark.hyperspace.explain.displayMode", "html");
spark.Conf().Set("spark.hyperspace.explain.displayMode.highlight.beginTag", "<b style=\"background:LightGreen\">");
spark.Conf().Set("spark.hyperspace.explain.displayMode.highlight.endTag", "</b>");
hyperspace.Explain(eqJoin, false, input => DisplayHTML(input));
结果：
有索引的计划
Project [empName#528, deptName#534]
+- SortMergeJoin [deptId#529], [deptId#533], Inner
:- *(1) Project [empName#528, deptId#529]
: +- *(1) Filter isnotnull(deptId#529)
: +- *(1) FileScan parquet [deptId#529,empName#528] Batched: true, Format: Parquet, Location: InMemoryFileIndex[abfss://datasets@hyperspacebenchmark.dfs.core.chinacloudapi.cn/hyperspaceon..., PartitionFilters: [], PushedFilters: [IsNotNull(deptId)], ReadSchema: struct
+- *(2) Project [deptId#533, deptName#534]
+- *(2) Filter isnotnull(deptId#533)
+- *(2) FileScan parquet [deptId#533,deptName#534] Batched: true, Format: Parquet, Location: InMemoryFileIndex[abfss://datasets@hyperspacebenchmark.dfs.core.chinacloudapi.cn/hyperspaceon..., PartitionFilters: [], PushedFilters: [IsNotNull(deptId)], ReadSchema: struct
无索引的计划
Project [empName#528, deptName#534]
+- SortMergeJoin [deptId#529], [deptId#533], Inner
:- *(2) Sort [deptId#529 ASC NULLS FIRST], false, 0
: +- Exchange hashpartitioning(deptId#529, 200)
: +- *(1) Project [empName#528, deptId#529]
: +- *(1) Filter isnotnull(deptId#529)
: +- *(1) FileScan parquet [empName#528,deptId#529] Batched: true, Format: Parquet, Location: InMemoryFileIndex[abfss://datasets@hyperspacebenchmark.dfs.core.chinacloudapi.cn/your-path/employees.parquet], PartitionFilters: [], PushedFilters: [IsNotNull(deptId)], ReadSchema: struct
+- *(4) Sort [deptId#533 ASC NULLS FIRST], false, 0
+- Exchange hashpartitioning(deptId#533, 200)
+- *(3) Project [deptId#533, deptName#534]
+- *(3) Filter isnotnull(deptId#533)
+- *(3) FileScan parquet [deptId#533,deptName#534] Batched: true, Format: Parquet, Location: InMemoryFileIndex[abfss://datasets@hyperspacebenchmark.dfs.core.chinacloudapi.cn/your-path/departments.parquet], PartitionFilters: [], PushedFilters: [IsNotNull(deptId)], ReadSchema: struct
使用的索引
deptIndex1:abfss://datasets@hyperspacebenchmark.dfs.core.chinacloudapi.cn/<container>/indexes/public/deptIndex1/v__=0
empIndex:abfss://datasets@hyperspacebenchmark.dfs.core.chinacloudapi.cn/<container>/indexes/public/empIndex/v__=0
刷新索引
如果创建索引的原始数据发生变化，索引将不再捕获最新数据状态。 可以使用 refreshIndex 命令刷新过时索引。 此命令可使索引完全被重建，并根据最新数据记录进行更新。 我们将向你展示如何在其他笔记本中以增量方式刷新索引。
以下两个单元显示了此情况的示例：
- 第一个单元在原始部门数据中又增加了两个部门。 它读取并输出了一份部门列表，以验证新部门是否正确添加。 输出显示总共六个部门：四个旧部门和两个新部门。 调用 refreshIndex 可更新“deptIndex1”，以便索引捕获新部门。
- 第二个单元运行范围选择查询示例。 结果现在应包含四个部门：两个部门是我们运行前一个查询时看到的部门，另外两个是我们添加的新部门。
特定索引刷新
val extraDepartments = Seq(
(50, "Inovation", "Seattle"),
(60, "Human Resources", "San Francisco"))
val extraDeptData: DataFrame = extraDepartments.toDF("deptId", "deptName", "location")
extraDeptData.write.mode("Append").parquet(dept_Location)
val deptDFrameUpdated: DataFrame = spark.read.parquet(dept_Location)
deptDFrameUpdated.show(10)
hyperspace.refreshIndex("deptIndex1")
extra_Departments = [(50, "Inovation", "Seattle"), (60, "Human Resources", "San Francisco")]
extra_departments_df = spark.createDataFrame(extra_Departments, dept_schema)
extra_departments_df.write.mode("Append").parquet(dept_Location)
dept_DFrame_Updated = spark.read.parquet(dept_Location)
dept_DFrame_Updated.show(10)
var extraDepartments = new List<GenericRow>()
{
new GenericRow(new object[] {50, "Inovation", "Seattle"}),
new GenericRow(new object[] {60, "Human Resources", "San Francisco"})
};
DataFrame extraDeptData = spark.CreateDataFrame(extraDepartments, departmentSchema);
extraDeptData.Write().Mode("Append").Parquet(dept_Location);
DataFrame deptDFrameUpdated = spark.Read().Parquet(dept_Location);
deptDFrameUpdated.Show(10);
hyperspace.RefreshIndex("deptIndex1");
结果：
extraDepartments: Seq[(Int, String, String)] = List((50,Inovation,Seattle), (60,Human Resources,San Francisco))
extraDeptData: org.apache.spark.sql.DataFrame = [deptId: int, deptName: string ... 1 more field]
deptDFrameUpdated: org.apache.spark.sql.DataFrame = [deptId: int, deptName: string ... 1 more field]
|deptId| deptName| location|
|------|---------------|-------------|
| 60|Human Resources|San Francisco|
| 10| Accounting| New York|
| 50| Inovation| Seattle|
| 40| Operations| Boston|
| 20| Research| Dallas|
| 30| Sales| Chicago|
范围选择
val newRangeFilter: DataFrame = deptDFrameUpdated.filter("deptId > 20").select("deptName")
newRangeFilter.show()
newRangeFilter.explain(true)
newRangeFilter = dept_DFrame_Updated.filter("deptId > 20").select("deptName")
newRangeFilter.show()
newRangeFilter.explain(True)
DataFrame newRangeFilter = deptDFrameUpdated.Filter("deptId > 20").Select("deptName");
newRangeFilter.Show();
newRangeFilter.Explain(true);
结果：
newRangeFilter: org.apache.spark.sql.DataFrame = [deptName: string]
| DeptName|
|---------------|
|Human Resources|
| Inovation|
| Operations|
| Sales|
== Parsed Logical Plan ==
'Project [unresolvedalias('deptName, None)]
+- Filter (deptId#674 > 20)
+- Relation[deptId#674,deptName#675,location#676] parquet
== Analyzed Logical Plan ==
deptName: string
Project [deptName#675]
+- Filter (deptId#674 > 20)
+- Relation[deptId#674,deptName#675,location#676] parquet
== Optimized Logical Plan ==
Project [deptName#675]
+- Filter (isnotnull(deptId#674) && (deptId#674 > 20))
+- Relation[deptId#674,deptName#675,location#676] parquet
== Physical Plan ==
*(1) Project [deptName#675]
+- *(1) Filter (isnotnull(deptId#674) && (deptId#674 > 20))
+- *(1) FileScan parquet [deptId#674,deptName#675] Batched: true, Format: Parquet, Location: InMemoryFileIndex[abfss://datasets@hyperspacebenchmark.dfs.core.chinacloudapi.cn/hyperspaceon..., PartitionFilters: [], PushedFilters: [IsNotNull(deptId), GreaterThan(deptId,20)], ReadSchema: struct<deptId:int,deptName:string>
可变数据集的混合扫描
通常，如果你的基础源数据追加了一些新文件或删除了现有文件，则你的索引将过时，Hyperspace 决定不使用它。 但是，有时你只想使用索引，而不必每次刷新索引。 这样做可能有多种原因：
- 你不想不断刷新索引，而是希望定期刷新索引，因为你最了解你的工作负载。
- 你只添加/删除了几个文件，不想等待另一个刷新工作完成。
为了让你仍然使用过时索引，Hyperspace 引入了混合扫描，这是一种新技术，允许用户使用过时或陈旧的索引（例如，基础源数据追加了一些新文件或删除了现有文件），而无需刷新索引。
为此，当你设置适当的配置以启用混合扫描时，Hyperspace 会修改查询计划，以利用以下更改：
- 追加的文件可以通过使用 Union 或 BucketUnion（用于联接）合并到索引数据。 如果需要，还可以在合并之前应用随机追加的数据。
- 删除的文件可以通过在索引数据的沿袭列上注入 Filter-NOT-IN 条件来处理，以便在查询时排除已删除文件中的索引行。
你可以在以下示例中检查查询计划的转换。
注意
目前，仅支持非分区数据的混合扫描。
用于追加的文件的混合扫描 - 非分区数据
在下面的示例中，将使用非分区数据。 在本示例中，我们希望联接索引可用于查询，并为追加的文件引入 BucketUnion。
val testData = Seq(
("orange", 3, "2020-10-01"),
("banana", 1, "2020-10-01"),
("carrot", 5, "2020-10-02"),
("beetroot", 12, "2020-10-02"),
("orange", 2, "2020-10-03"),
("banana", 11, "2020-10-03"),
("carrot", 3, "2020-10-03"),
("beetroot", 2, "2020-10-04"),
("cucumber", 7, "2020-10-05"),
("pepper", 20, "2020-10-06")
).toDF("name", "qty", "date")
val testDataLocation = s"$dataPath/productTable"
testData.write.mode("overwrite").parquet(testDataLocation)
val testDF = spark.read.parquet(testDataLocation)
testdata = [
("orange", 3, "2020-10-01"),
("banana", 1, "2020-10-01"),
("carrot", 5, "2020-10-02"),
("beetroot", 12, "2020-10-02"),
("orange", 2, "2020-10-03"),
("banana", 11, "2020-10-03"),
("carrot", 3, "2020-10-03"),
("beetroot", 2, "2020-10-04"),
("cucumber", 7, "2020-10-05"),
("pepper", 20, "2020-10-06")
]
testdata_location = data_path + "/productTable"
from pyspark.sql.types import StructField, StructType, StringType, IntegerType
testdata_schema = StructType([
StructField('name', StringType(), True),
StructField('qty', IntegerType(), True),
StructField('date', StringType(), True)])
test_df = spark.createDataFrame(testdata, testdata_schema)
test_df.write.mode("overwrite").parquet(testdata_location)
test_df = spark.read.parquet(testdata_location)
using Microsoft.Spark.Sql.Types;
var products = new List<GenericRow>() {
new GenericRow(new object[] {"orange", 3, "2020-10-01"}),
new GenericRow(new object[] {"banana", 1, "2020-10-01"}),
new GenericRow(new object[] {"carrot", 5, "2020-10-02"}),
new GenericRow(new object[] {"beetroot", 12, "2020-10-02"}),
new GenericRow(new object[] {"orange", 2, "2020-10-03"}),
new GenericRow(new object[] {"banana", 11, "2020-10-03"}),
new GenericRow(new object[] {"carrot", 3, "2020-10-03"}),
new GenericRow(new object[] {"beetroot", 2, "2020-10-04"}),
new GenericRow(new object[] {"cucumber", 7, "2020-10-05"}),
new GenericRow(new object[] {"pepper", 20, "2020-10-06"})
};
var productsSchema = new StructType(new List<StructField>()
{
new StructField("name", new StringType()),
new StructField("qty", new IntegerType()),
new StructField("date", new StringType())
});
DataFrame testData = spark.CreateDataFrame(products, productsSchema);
string testDataLocation = $"{dataPath}/productTable";
testData.Write().Mode("overwrite").Parquet(testDataLocation);
// CREATE INDEX
hyperspace.createIndex(testDF, IndexConfig("productIndex2", Seq("name"), Seq("date", "qty")))
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", -1)
val filter1 = testDF.filter("name = 'banana'")
val filter2 = testDF.filter("qty > 10")
val query = filter1.join(filter2, "name")
// Check Join index rule is applied properly.
hyperspace.explain(query)(displayHTML(_))
# CREATE INDEX
hyperspace.createIndex(test_df, IndexConfig("productIndex2", ["name"], ["date", "qty"]))
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", -1)
filter1 = test_df.filter("name = 'banana'")
filter2 = test_df.filter("qty > 10")
query = filter1.join(filter2, "name")
# Check Join index rule is applied properly.
hyperspace.explain(query, True, displayHTML)
// CREATE INDEX
DataFrame testDF = spark.Read().Parquet(testDataLocation);
var productIndex2Config = new IndexConfig("productIndex", new string[] {"name"}, new string[] {"date", "qty"});
hyperspace.CreateIndex(testDF, productIndex2Config);
// Check Join index rule is applied properly.
DataFrame filter1 = testDF.Filter("name = 'banana'");
DataFrame filter2 = testDF.Filter("qty > 10");
DataFrame query = filter1.Join(filter2, filter1.Col("name") == filter2.Col("name"));
query.Show();
hyperspace.Explain(query, true, input => DisplayHTML(input));
结果为：
=============================================================
Plan with indexes:
=============================================================
Project [name#607, qty#608, date#609, qty#632, date#633]
+- SortMergeJoin [name#607], [name#631], Inner
:- *(1) Project [name#607, qty#608, date#609]
: +- *(1) Filter (isnotnull(name#607) && (name#607 = banana))
: +- *(1) FileScan parquet [name#607,date#609,qty#608] Batched: true, Format: Parquet, Location: InMemoryFileIndex[abfss://datasets@hyperspacebenchmark.dfs.core.chinacloudapi.cn/hyperspace/indexes-777519/p..., PartitionFilters: [], PushedFilters: [IsNotNull(name), EqualTo(name,banana)], ReadSchema: struct, SelectedBucketsCount: 1 out of 200
+- *(2) Project [name#631, qty#632, date#633]
+- *(2) Filter (((isnotnull(qty#632) && (qty#632 > 10)) && isnotnull(name#631)) && (name#631 = banana))
+- *(2) FileScan parquet [name#631,date#633,qty#632] Batched: true, Format: Parquet, Location: InMemoryFileIndex[abfss://datasets@hyperspacebenchmark.dfs.core.chinacloudapi.cn/hyperspace/indexes-777519/p..., PartitionFilters: [], PushedFilters: [IsNotNull(qty), GreaterThan(qty,10), IsNotNull(name), EqualTo(name,banana)], ReadSchema: struct, SelectedBucketsCount: 1 out of 200
=============================================================
Plan without indexes:
=============================================================
Project [name#607, qty#608, date#609, qty#632, date#633]
+- SortMergeJoin [name#607], [name#631], Inner
:- *(2) Sort [name#607 ASC NULLS FIRST], false, 0
: +- Exchange hashpartitioning(name#607, 200), [id=#453]
: +- *(1) Project [name#607, qty#608, date#609]
: +- *(1) Filter (isnotnull(name#607) && (name#607 = banana))
: +- *(1) FileScan parquet [name#607,qty#608,date#609] Batched: true, Format: Parquet, Location: InMemoryFileIndex[abfss://datasets@hyperspacebenchmark.dfs.core.chinacloudapi.cn/hyperspace/data-777519/prod..., PartitionFilters: [], PushedFilters: [IsNotNull(name), EqualTo(name,banana)], ReadSchema: struct
+- *(4) Sort [name#631 ASC NULLS FIRST], false, 0
+- Exchange hashpartitioning(name#631, 200), [id=#459]
+- *(3) Project [name#631, qty#632, date#633]
+- *(3) Filter (((isnotnull(qty#632) && (qty#632 > 10)) && isnotnull(name#631)) && (name#631 = banana))
+- *(3) FileScan parquet [name#631,qty#632,date#633] Batched: true, Format: Parquet, Location: InMemoryFileIndex[abfss://datasets@hyperspacebenchmark.dfs.core.chinacloudapi.cn/hyperspace/data-777519/prod..., PartitionFilters: [], PushedFilters: [IsNotNull(qty), GreaterThan(qty,10), IsNotNull(name), EqualTo(name,banana)], ReadSchema: struct
=============================================================
Indexes used:
=============================================================
productIndex2:abfss://datasets@hyperspacebenchmark.dfs.core.chinacloudapi.cn/hyperspace/indexes-777519/productIndex2/v__=0
// Append new files.
val appendData = Seq(
("orange", 13, "2020-11-01"),
("banana", 5, "2020-11-01")).toDF("name", "qty", "date")
appendData.write.mode("append").parquet(testDataLocation)
# Append new files.
append_data = [
("orange", 13, "2020-11-01"),
("banana", 5, "2020-11-01")
]
append_df = spark.createDataFrame(append_data, testdata_schema)
append_df.write.mode("append").parquet(testdata_location)
// Append new files.
var appendProducts = new List<GenericRow>()
{
new GenericRow(new object[] {"orange", 13, "2020-11-01"}),
new GenericRow(new object[] {"banana", 5, "2020-11-01"})
};
DataFrame appendData = spark.CreateDataFrame(appendProducts, productsSchema);
appendData.Write().Mode("Append").Parquet(testDataLocation);
默认情况下，混合扫描处于禁用状态。 因此，你将看到，由于我们追加了新数据，Hyperspace 将决定不使用索引。
在输出中，你将看不到计划差异（因此，不会突出显示）。
// Hybrid Scan configs are false by default.
spark.conf.set("spark.hyperspace.index.hybridscan.enabled", "false")
spark.conf.set("spark.hyperspace.index.hybridscan.delete.enabled", "false")
val testDFWithAppend = spark.read.parquet(testDataLocation)
val filter1 = testDFWithAppend.filter("name = 'banana'")
val filter2 = testDFWithAppend.filter("qty > 10")
val query = filter1.join(filter2, "name")
hyperspace.explain(query)(displayHTML(_))
query.show
# Hybrid Scan configs are false by default.
spark.conf.set("spark.hyperspace.index.hybridscan.enabled", "false")
spark.conf.set("spark.hyperspace.index.hybridscan.delete.enabled", "false")
test_df_with_append = spark.read.parquet(testdata_location)
filter1 = test_df_with_append.filter("name = 'banana'")
filter2 = test_df_with_append.filter("qty > 10")
query = filter1.join(filter2, "name")
hyperspace.explain(query, True, displayHTML)
query.show()
// Hybrid Scan configs are false by default.
spark.Conf().Set("spark.hyperspace.index.hybridscan.enabled", "false");
spark.Conf().Set("spark.hyperspace.index.hybridscan.delete.enabled", "false");
DataFrame testDFWithAppend = spark.Read().Parquet(testDataLocation);
DataFrame filter1 = testDFWithAppend.Filter("name = 'banana'");
DataFrame filter2 = testDFWithAppend.Filter("qty > 10");
DataFrame query = filter1.Join(filter2, filter1.Col("name") == filter2.Col("name"));
query.Show();
hyperspace.Explain(query, true, input => DisplayHTML(input));
结果为：
=============================================================
Plan with indexes:
=============================================================
Project [name#678, qty#679, date#680, qty#685, date#686]
+- SortMergeJoin [name#678], [name#684], Inner
:- *(2) Sort [name#678 ASC NULLS FIRST], false, 0
: +- Exchange hashpartitioning(name#678, 200), [id=#589]
: +- *(1) Project [name#678, qty#679, date#680]
: +- *(1) Filter (isnotnull(name#678) && (name#678 = banana))
: +- *(1) FileScan parquet [name#678,qty#679,date#680] Batched: true, Format: Parquet, Location: InMemoryFileIndex[abfss://datasets@hyperspacebenchmark.dfs.core.chinacloudapi.cn/hyperspace/data-777519/prod..., PartitionFilters: [], PushedFilters: [IsNotNull(name), EqualTo(name,banana)], ReadSchema: struct
+- *(4) Sort [name#684 ASC NULLS FIRST], false, 0
+- Exchange hashpartitioning(name#684, 200), [id=#595]
+- *(3) Project [name#684, qty#685, date#686]
+- *(3) Filter (((isnotnull(qty#685) && (qty#685 > 10)) && (name#684 = banana)) && isnotnull(name#684))
+- *(3) FileScan parquet [name#684,qty#685,date#686] Batched: true, Format: Parquet, Location: InMemoryFileIndex[abfss://datasets@hyperspacebenchmark.dfs.core.chinacloudapi.cn/hyperspace/data-777519/prod..., PartitionFilters: [], PushedFilters: [IsNotNull(qty), GreaterThan(qty,10), EqualTo(name,banana), IsNotNull(name)], ReadSchema: struct
=============================================================
Plan without indexes:
=============================================================
Project [name#678, qty#679, date#680, qty#685, date#686]
+- SortMergeJoin [name#678], [name#684], Inner
:- *(2) Sort [name#678 ASC NULLS FIRST], false, 0
: +- Exchange hashpartitioning(name#678, 200), [id=#536]
: +- *(1) Project [name#678, qty#679, date#680]
: +- *(1) Filter (isnotnull(name#678) && (name#678 = banana))
: +- *(1) FileScan parquet [name#678,qty#679,date#680] Batched: true, Format: Parquet, Location: InMemoryFileIndex[abfss://datasets@hyperspacebenchmark.dfs.core.chinacloudapi.cn/hyperspace/data-777519/prod..., PartitionFilters: [], PushedFilters: [IsNotNull(name), EqualTo(name,banana)], ReadSchema: struct
+- *(4) Sort [name#684 ASC NULLS FIRST], false, 0
+- Exchange hashpartitioning(name#684, 200), [id=#542]
+- *(3) Project [name#684, qty#685, date#686]
+- *(3) Filter (((isnotnull(qty#685) && (qty#685 > 10)) && (name#684 = banana)) && isnotnull(name#684))
+- *(3) FileScan parquet [name#684,qty#685,date#686] Batched: true, Format: Parquet, Location: InMemoryFileIndex[abfss://datasets@hyperspacebenchmark.dfs.core.chinacloudapi.cn/hyperspace/data-777519/prod..., PartitionFilters: [], PushedFilters: [IsNotNull(qty), GreaterThan(qty,10), EqualTo(name,banana), IsNotNull(name)], ReadSchema: struct
+------+---+----------+---+----------+
| name|qty| date|qty| date|
+------+---+----------+---+----------+
|banana| 11|2020-10-03| 11|2020-10-03|
|banana| 5|2020-11-01| 11|2020-10-03|
|banana| 1|2020-10-01| 11|2020-10-03|
+------+---+----------+---+----------
启用混合扫描
在有索引的计划中，你可以看到仅追加的文件需要的 Exchange 哈希分区，以便我们仍然可以使用带有追加的文件的“随机”索引数据。 BucketUnion 用于将“随机”追加的文件和索引数据合并在一起。
// Enable Hybrid Scan config. "delete" config is not necessary since we only appended data.
spark.conf.set("spark.hyperspace.index.hybridscan.enabled", "true")
spark.enableHyperspace
// Need to redefine query to recalculate the query plan.
val query = filter1.join(filter2, "name")
hyperspace.explain(query)(displayHTML(_))
query.show
# Enable Hybrid Scan config. "delete" config is not necessary.
spark.conf.set("spark.hyperspace.index.hybridscan.enabled", "true")
# Need to redefine query to recalculate the query plan.
query = filter1.join(filter2, "name")
hyperspace.explain(query, True, displayHTML)
query.show()
// Enable Hybrid Scan config. "delete" config is not necessary.
spark.Conf().Set("spark.hyperspace.index.hybridscan.enabled", "true");
spark.EnableHyperspace();
// Need to redefine query to recalculate the query plan.
DataFrame query = filter1.Join(filter2, filter1.Col("name") == filter2.Col("name"));
query.Show();
hyperspace.Explain(query, true, input => DisplayHTML(input));
结果为：
=============================================================
Plan with indexes:
=============================================================
Project [name#678, qty#679, date#680, qty#732, date#733]
+- SortMergeJoin [name#678], [name#731], Inner
:- *(3) Sort [name#678 ASC NULLS FIRST], false, 0
: +- BucketUnion 200 buckets, bucket columns: [name]
: :- *(1) Project [name#678, qty#679, date#680]
: : +- *(1) Filter (isnotnull(name#678) && (name#678 = banana))
: : +- *(1) FileScan parquet [name#678,date#680,qty#679] Batched: true, Format: Parquet, Location: InMemoryFileIndex[abfss://datasets@hyperspacebenchmark.dfs.core.chinacloudapi.cn/hyperspace/indexes-777519/p..., PartitionFilters: [], PushedFilters: [IsNotNull(name), EqualTo(name,banana)], ReadSchema: struct, SelectedBucketsCount: 1 out of 200
: +- Exchange hashpartitioning(name#678, 200), [id=#775]
: +- *(2) Project [name#678, qty#679, date#680]
: +- *(2) Filter (isnotnull(name#678) && (name#678 = banana))
: +- *(2) FileScan parquet [name#678,date#680,qty#679] Batched: true, Format: Parquet, Location: InMemoryFileIndex[abfss://datasets@hyperspacebenchmark.dfs.core.chinacloudapi.cn/hyperspace/data-777519/prod..., PartitionFilters: [], PushedFilters: [IsNotNull(name), EqualTo(name,banana)], ReadSchema: struct
+- *(6) Sort [name#731 ASC NULLS FIRST], false, 0
+- BucketUnion 200 buckets, bucket columns: [name]
:- *(4) Project [name#731, qty#732, date#733]
: +- *(4) Filter (((isnotnull(qty#732) && (qty#732 > 10)) && isnotnull(name#731)) && (name#731 = banana))
: +- *(4) FileScan parquet [name#731,date#733,qty#732] Batched: true, Format: Parquet, Location: InMemoryFileIndex[abfss://datasets@hyperspacebenchmark.dfs.core.chinacloudapi.cn/hyperspace/indexes-777519/p..., PartitionFilters: [], PushedFilters: [IsNotNull(qty), GreaterThan(qty,10), IsNotNull(name), EqualTo(name,banana)], ReadSchema: struct, SelectedBucketsCount: 1 out of 200
+- Exchange hashpartitioning(name#731, 200), [id=#783]
+- *(5) Project [name#731, qty#732, date#733]
+- *(5) Filter (((isnotnull(qty#732) && (qty#732 > 10)) && isnotnull(name#731)) && (name#731 = banana))
+- *(5) FileScan parquet [name#731,date#733,qty#732] Batched: true, Format: Parquet, Location: InMemoryFileIndex[abfss://datasets@hyperspacebenchmark.dfs.core.chinacloudapi.cn/hyperspace/data-777519/prod..., PartitionFilters: [], PushedFilters: [IsNotNull(qty), GreaterThan(qty,10), IsNotNull(name), EqualTo(name,banana)], ReadSchema: struct
=============================================================
Plan without indexes:
=============================================================
Project [name#678, qty#679, date#680, qty#732, date#733]
+- SortMergeJoin [name#678], [name#731], Inner
:- *(2) Sort [name#678 ASC NULLS FIRST], false, 0
: +- Exchange hashpartitioning(name#678, 200), [id=#701]
: +- *(1) Project [name#678, qty#679, date#680]
: +- *(1) Filter (isnotnull(name#678) && (name#678 = banana))
: +- *(1) FileScan parquet [name#678,qty#679,date#680] Batched: true, Format: Parquet, Location: InMemoryFileIndex[abfss://datasets@hyperspacebenchmark.dfs.core.chinacloudapi.cn/hyperspace/data-777519/prod..., PartitionFilters: [], PushedFilters: [IsNotNull(name), EqualTo(name,banana)], ReadSchema: struct
+- *(4) Sort [name#731 ASC NULLS FIRST], false, 0
+- Exchange hashpartitioning(name#731, 200), [id=#707]
+- *(3) Project [name#731, qty#732, date#733]
+- *(3) Filter (((isnotnull(qty#732) && (qty#732 > 10)) && isnotnull(name#731)) && (name#731 = banana))
+- *(3) FileScan parquet [name#731,qty#732,date#733] Batched: true, Format: Parquet, Location: InMemoryFileIndex[abfss://datasets@hyperspacebenchmark.dfs.core.chinacloudapi.cn/hyperspace/data-777519/prod..., PartitionFilters: [], PushedFilters: [IsNotNull(qty), GreaterThan(qty,10), IsNotNull(name), EqualTo(name,banana)], ReadSchema: struct
=============================================================
Indexes used:
=============================================================
productIndex2:abfss://datasets@hyperspacebenchmark.dfs.core.chinacloudapi.cn/hyperspace/indexes-777519/productIndex2/v__=0
+------+---+----------+---+----------+
| name|qty| date|qty| date|
+------+---+----------+---+----------+
|banana| 1|2020-10-01| 11|2020-10-03|
|banana| 11|2020-10-03| 11|2020-10-03|
|banana| 5|2020-11-01| 11|2020-10-03|
+------+---+----------+---+----------+
增量索引刷新
当你准备好更新索引但不想重新生成整个索引时，Hyperspace 支持使用 hs.refreshIndex("name", "incremental") API 以增量方式更新索引。 这将消除从头开始全面重新生成索引的需要，而是利用以前创建的索引文件，以及仅更新新添加的数据上的索引。
当然，请务必定期使用补充的 optimizeIndex API（如下所示），以确保你不会看到性能回归。 假设你添加/删除的数据小于原始数据集的 10%，则建议每调用 refreshIndex(..., "incremental") 10 次至少调用一次优化。 例如，如果你的原始数据集为 100 GB，并且已以 1 GB 的增量/减量添加/删除数据，则可以在调用 optimizeIndex 之前调用 refreshIndex 10 次。 请注意，此示例用于演示，你必须根据工作负载对此进行调整。
在下面的示例中，请注意，使用索引时在查询计划中添加了排序节点。 这是因为在追加的数据文件上创建了部分索引，从而导致 Spark 引入 Sort。 另请注意，Shuffle（即 Exchange）仍会从计划中消除，从而为你提供适当的加速。
def query(): DataFrame = {
val testDFWithAppend = spark.read.parquet(testDataLocation)
val filter1 = testDFWithAppend.filter("name = 'banana'")
val filter2 = testDFWithAppend.filter("qty > 10")
filter1.join(filter2, "name")
}
hyperspace.refreshIndex("productIndex2", "incremental")
hyperspace.explain(query())(displayHTML(_))
query().show
def query():
test_df_with_append = spark.read.parquet(testdata_location)
filter1 = test_df_with_append.filter("name = 'banana'")
filter2 = test_df_with_append.filter("qty > 10")
return filter1.join(filter2, "name")
hyperspace.refreshIndex("productIndex2", "incremental")
hyperspace.explain(query(), True, displayHTML)
query().show()
结果为：
=============================================================
Plan with indexes:
=============================================================
Project [name#820, qty#821, date#822, qty#827, date#828]
+- SortMergeJoin [name#820], [name#826], Inner
:- *(1) Sort [name#820 ASC NULLS FIRST], false, 0
: +- *(1) Project [name#820, qty#821, date#822]
: +- *(1) Filter (isnotnull(name#820) && (name#820 = banana))
: +- *(1) FileScan parquet [name#820,date#822,qty#821] Batched: true, Format: Parquet, Location: InMemoryFileIndex[abfss://datasets@hyperspacebenchmark.dfs.core.chinacloudapi.cn/hyperspace/indexes-777519/p..., PartitionFilters: [], PushedFilters: [IsNotNull(name), EqualTo(name,banana)], ReadSchema: struct, SelectedBucketsCount: 1 out of 200
+- *(2) Sort [name#826 ASC NULLS FIRST], false, 0
+- *(2) Project [name#826, qty#827, date#828]
+- *(2) Filter (((isnotnull(qty#827) && (qty#827 > 10)) && (name#826 = banana)) && isnotnull(name#826))
+- *(2) FileScan parquet [name#826,date#828,qty#827] Batched: true, Format: Parquet, Location: InMemoryFileIndex[abfss://datasets@hyperspacebenchmark.dfs.core.chinacloudapi.cn/hyperspace/indexes-777519/p..., PartitionFilters: [], PushedFilters: [IsNotNull(qty), GreaterThan(qty,10), EqualTo(name,banana), IsNotNull(name)], ReadSchema: struct, SelectedBucketsCount: 1 out of 200
=============================================================
Plan without indexes:
=============================================================
Project [name#820, qty#821, date#822, qty#827, date#828]
+- SortMergeJoin [name#820], [name#826], Inner
:- *(2) Sort [name#820 ASC NULLS FIRST], false, 0
: +- Exchange hashpartitioning(name#820, 200), [id=#927]
: +- *(1) Project [name#820, qty#821, date#822]
: +- *(1) Filter (isnotnull(name#820) && (name#820 = banana))
: +- *(1) FileScan parquet [name#820,qty#821,date#822] Batched: true, Format: Parquet, Location: InMemoryFileIndex[abfss://datasets@hyperspacebenchmark.dfs.core.chinacloudapi.cn/hyperspace/data-777519/prod..., PartitionFilters: [], PushedFilters: [IsNotNull(name), EqualTo(name,banana)], ReadSchema: struct
+- *(4) Sort [name#826 ASC NULLS FIRST], false, 0
+- Exchange hashpartitioning(name#826, 200), [id=#933]
+- *(3) Project [name#826, qty#827, date#828]
+- *(3) Filter (((isnotnull(qty#827) && (qty#827 > 10)) && (name#826 = banana)) && isnotnull(name#826))
+- *(3) FileScan parquet [name#826,qty#827,date#828] Batched: true, Format: Parquet, Location: InMemoryFileIndex[abfss://datasets@hyperspacebenchmark.dfs.core.chinacloudapi.cn/hyperspace/data-777519/prod..., PartitionFilters: [], PushedFilters: [IsNotNull(qty), GreaterThan(qty,10), EqualTo(name,banana), IsNotNull(name)], ReadSchema: struct
+------+---+----------+---+----------+
| name|qty| date|qty| date|
+------+---+----------+---+----------+
|banana| 1|2020-10-01| 11|2020-10-03|
|banana| 11|2020-10-03| 11|2020-10-03|
|banana| 5|2020-11-01| 11|2020-10-03|
+------+---+----------+---+----------+
优化索引布局
在新追加的数据上多次调用增量刷新后（例如，如果用户以小批量或在流式处理情况下写入到数据），索引文件的数量往往会变大，从而影响索引的性能（大量小文件问题）。 Hyperspace 提供 hyperspace.optimizeIndex("indexName") API 以优化索引布局并减少大文件问题。
在下面的计划中，请注意，Hyperspace 已删除查询计划中的其他排序节点。 优化可以帮助避免对仅包含一个文件的任何索引存储桶进行排序。 但是，只有当所有索引存储桶在 optimizeIndex 之后每个存储桶最多拥有 1 个文件时，才会如此。
// Append some more data and call refresh again.
val appendData = Seq(
("orange", 13, "2020-11-01"),
("banana", 5, "2020-11-01")).toDF("name", "qty", "date")
appendData.write.mode("append").parquet(testDataLocation)
hyperspace.refreshIndex("productIndex2", "incremental")
# Append some more data and call refresh again.
append_data = [
("orange", 13, "2020-11-01"),
("banana", 5, "2020-11-01")
]
append_df = spark.createDataFrame(append_data, testdata_schema)
append_df.write.mode("append").parquet(testdata_location)
hyperspace.refreshIndex("productIndex2", "incremental"
// Call optimize. Ensure that Sort is removed after optimization (This is possible here because after optimize, in this case, every bucket contains only 1 file.).
hyperspace.optimizeIndex("productIndex2")
hyperspace.explain(query())(displayHTML(_))
# Call optimize. Ensure that Sort is removed after optimization (This is possible here because after optimize, in this case, every bucket contains only 1 file.).
hyperspace.optimizeIndex("productIndex2")
hyperspace.explain(query(), True, displayHTML)
结果为：
=============================================================
Plan with indexes:
=============================================================
Project [name#954, qty#955, date#956, qty#961, date#962]
+- SortMergeJoin [name#954], [name#960], Inner
:- *(1) Project [name#954, qty#955, date#956]
: +- *(1) Filter (isnotnull(name#954) && (name#954 = banana))
: +- *(1) FileScan parquet [name#954,date#956,qty#955] Batched: true, Format: Parquet, Location: InMemoryFileIndex[abfss://datasets@hyperspacebenchmark.dfs.core.chinacloudapi.cn/hyperspace/indexes-777519/p..., PartitionFilters: [], PushedFilters: [IsNotNull(name), EqualTo(name,banana)], ReadSchema: struct, SelectedBucketsCount: 1 out of 200
+- *(2) Project [name#960, qty#961, date#962]
+- *(2) Filter (((isnotnull(qty#961) && (qty#961 > 10)) && isnotnull(name#960)) && (name#960 = banana))
+- *(2) FileScan parquet [name#960,date#962,qty#961] Batched: true, Format: Parquet, Location: InMemoryFileIndex[abfss://datasets@hyperspacebenchmark.dfs.core.chinacloudapi.cn/hyperspace/indexes-777519/p..., PartitionFilters: [], PushedFilters: [IsNotNull(qty), GreaterThan(qty,10), IsNotNull(name), EqualTo(name,banana)], ReadSchema: struct, SelectedBucketsCount: 1 out of 200
=============================================================
Plan without indexes:
=============================================================
Project [name#954, qty#955, date#956, qty#961, date#962]
+- SortMergeJoin [name#954], [name#960], Inner
:- *(2) Sort [name#954 ASC NULLS FIRST], false, 0
: +- Exchange hashpartitioning(name#954, 200), [id=#1070]
: +- *(1) Project [name#954, qty#955, date#956]
: +- *(1) Filter (isnotnull(name#954) && (name#954 = banana))
: +- *(1) FileScan parquet [name#954,qty#955,date#956] Batched: true, Format: Parquet, Location: InMemoryFileIndex[abfss://datasets@hyperspacebenchmark.dfs.core.chinacloudapi.cn/hyperspace/data-777519/prod..., PartitionFilters: [], PushedFilters: [IsNotNull(name), EqualTo(name,banana)], ReadSchema: struct
+- *(4) Sort [name#960 ASC NULLS FIRST], false, 0
+- Exchange hashpartitioning(name#960, 200), [id=#1076]
+- *(3) Project [name#960, qty#961, date#962]
+- *(3) Filter (((isnotnull(qty#961) && (qty#961 > 10)) && isnotnull(name#960)) && (name#960 = banana))
+- *(3) FileScan parquet [name#960,qty#961,date#962] Batched: true, Format: Parquet, Location: InMemoryFileIndex[abfss://datasets@hyperspacebenchmark.dfs.core.chinacloudapi.cn/hyperspace/data-777519/prod..., PartitionFilters: [], PushedFilters: [IsNotNull(qty), GreaterThan(qty,10), IsNotNull(name), EqualTo(name,banana)], ReadSchema: struct
=============================================================
Indexes used:
=============================================================
productIndex2:abfss://datasets@hyperspacebenchmark.dfs.core.chinacloudapi.cn/hyperspace/indexes-777519/productIndex2/v__=3
优化模式
优化的默认模式是“快速”模式，在这种模式下，会选择小于预定义阈值的文件进行优化。 为了最大限度地发挥优化的效果，Hyperspace 允许使用另一种优化模式，即“完整”，如下所示。 此模式选择所有索引文件进行优化，而不论其文件大小如何，并创建最可能的索引布局。 此模式也比默认优化模式慢，因为需要处理更多数据。
hyperspace.optimizeIndex("productIndex2", "full")
hyperspace.explain(query())(displayHTML(_))
hyperspace.optimizeIndex("productIndex2", "full")
hyperspace.explain(query(), True, displayHTML)
结果为：
=============================================================
Plan with indexes:
=============================================================
Project [name#1000, qty#1001, date#1002, qty#1007, date#1008]
+- SortMergeJoin [name#1000], [name#1006], Inner
:- *(1) Project [name#1000, qty#1001, date#1002]
: +- *(1) Filter (isnotnull(name#1000) && (name#1000 = banana))
: +- *(1) FileScan parquet [name#1000,date#1002,qty#1001] Batched: true, Format: Parquet, Location: InMemoryFileIndex[abfss://datasets@hyperspacebenchmark.dfs.core.chinacloudapi.cn/hyperspace/indexes-777519/p..., PartitionFilters: [], PushedFilters: [IsNotNull(name), EqualTo(name,banana)], ReadSchema: struct, SelectedBucketsCount: 1 out of 200
+- *(2) Project [name#1006, qty#1007, date#1008]
+- *(2) Filter (((isnotnull(qty#1007) && (qty#1007 > 10)) && isnotnull(name#1006)) && (name#1006 = banana))
+- *(2) FileScan parquet [name#1006,date#1008,qty#1007] Batched: true, Format: Parquet, Location: InMemoryFileIndex[abfss://datasets@hyperspacebenchmark.dfs.core.chinacloudapi.cn/hyperspace/indexes-777519/p..., PartitionFilters: [], PushedFilters: [IsNotNull(qty), GreaterThan(qty,10), IsNotNull(name), EqualTo(name,banana)], ReadSchema: struct, SelectedBucketsCount: 1 out of 200
=============================================================
Plan without indexes:
=============================================================
Project [name#1000, qty#1001, date#1002, qty#1007, date#1008]
+- SortMergeJoin [name#1000], [name#1006], Inner
:- *(2) Sort [name#1000 ASC NULLS FIRST], false, 0
: +- Exchange hashpartitioning(name#1000, 200), [id=#1160]
: +- *(1) Project [name#1000, qty#1001, date#1002]
: +- *(1) Filter (isnotnull(name#1000) && (name#1000 = banana))
: +- *(1) FileScan parquet [name#1000,qty#1001,date#1002] Batched: true, Format: Parquet, Location: InMemoryFileIndex[abfss://datasets@hyperspacebenchmark.dfs.core.chinacloudapi.cn/hyperspace/data-777519/prod..., PartitionFilters: [], PushedFilters: [IsNotNull(name), EqualTo(name,banana)], ReadSchema: struct
+- *(4) Sort [name#1006 ASC NULLS FIRST], false, 0
+- Exchange hashpartitioning(name#1006, 200), [id=#1166]
+- *(3) Project [name#1006, qty#1007, date#1008]
+- *(3) Filter (((isnotnull(qty#1007) && (qty#1007 > 10)) && isnotnull(name#1006)) && (name#1006 = banana))
+- *(3) FileScan parquet [name#1006,qty#1007,date#1008] Batched: true, Format: Parquet, Location: InMemoryFileIndex[abfss://datasets@hyperspacebenchmark.dfs.core.chinacloudapi.cn/hyperspace/data-777519/prod..., PartitionFilters: [], PushedFilters: [IsNotNull(qty), GreaterThan(qty,10), IsNotNull(name), EqualTo(name,banana)], ReadSchema: struct
=============================================================
Indexes used:
=============================================================
productIndex2:abfss://datasets@hyperspacebenchmark.dfs.core.chinacloudapi.cn/hyperspace/indexes-777519/productIndex2/v__=4