本指南通过使用spark-shell简要介绍了Hudi功能。使用Spark数据源，我们将通过代码段展示如何插入和更新的Hudi默认存储类型数据集： 写时复制。每次写操作之后，我们还将展示如何读取快照和增量读取数据。

设置spark-shell

Hudi适用于Spark-2.x版本。您可以按照此处的说明设置spark。 在提取的目录中，使用spark-shell运行Hudi：

  1. bin/spark-shell --packages org.apache.hudi:hudi-spark-bundle:0.5.0-incubating --conf 'spark.serializer=org.apache.spark.serializer.KryoSerializer'

设置表名、基本路径和数据生成器来为本指南生成记录。

  1. import org.apache.hudi.QuickstartUtils._
  2. import scala.collection.JavaConversions._
  3. import org.apache.spark.sql.SaveMode._
  4. import org.apache.hudi.DataSourceReadOptions._
  5. import org.apache.hudi.DataSourceWriteOptions._
  6. import org.apache.hudi.config.HoodieWriteConfig._
  8. val tableName = "hudi_cow_table"
  9. val basePath = "file:///tmp/hudi_cow_table"
  10. val dataGen = new DataGenerator

数据生成器 可以基于行程样本模式 生成插入和更新的样本。

插入数据 {#inserts}

生成一些新的行程样本，将其加载到DataFrame中，然后将DataFrame写入Hudi数据集中，如下所示。

  1. val inserts = convertToStringList(dataGen.generateInserts(10))
  2. val df = spark.read.json(spark.sparkContext.parallelize(inserts, 2))
  3. df.write.format("org.apache.hudi").
  4.     options(getQuickstartWriteConfigs).
  5.     option(PRECOMBINE_FIELD_OPT_KEY, "ts").
  6.     option(RECORDKEY_FIELD_OPT_KEY, "uuid").
  7.     option(PARTITIONPATH_FIELD_OPT_KEY, "partitionpath").
  8.     option(TABLE_NAME, tableName).
  9.     mode(Overwrite).
  10.     save(basePath);

mode(Overwrite)覆盖并重新创建数据集(如果已经存在)。 您可以检查在/tmp/hudi_cow_table/<region>/<country>/<city>/下生成的数据。我们提供了一个记录键 (schema中的uuid)，分区字段(region/county/city）和组合逻辑(schema中的ts) 以确保行程记录在每个分区中都是唯一的。更多信息请参阅 对Hudi中的数据进行建模-HowdoImodelthedatastoredinHudi?)， 有关将数据提取到Hudi中的方法的信息，请参阅写入Hudi数据集。 这里我们使用默认的写操作：插入更新。 如果您的工作负载没有更新，也可以使用更快的插入或批量插入操作。 想了解更多信息，请参阅写操作

查询数据 {#query}

将数据文件加载到DataFrame中。

  1. val roViewDF = spark.
  2.     read.
  3.     format("org.apache.hudi").
  4.     load(basePath + "/*/*/*/*")
  5. roViewDF.registerTempTable("hudi_ro_table")
  6. spark.sql("select fare, begin_lon, begin_lat, ts from  hudi_ro_table where fare > 20.0").show()
  7. spark.sql("select _hoodie_commit_time, _hoodie_record_key, _hoodie_partition_path, rider, driver, fare from  hudi_ro_table").show()

该查询提供已提取数据的读取优化视图。由于我们的分区路径(region/country/city)是嵌套的3个级别 从基本路径开始，我们使用了load(basePath + "/*/*/*/*")。 有关支持的所有存储类型和视图的更多信息，请参考存储类型和视图。

更新数据 {#updates}

这类似于插入新数据。使用数据生成器生成对现有行程的更新，加载到DataFrame中并将DataFrame写入hudi数据集。

  1. val updates = convertToStringList(dataGen.generateUpdates(10))
  2. val df = spark.read.json(spark.sparkContext.parallelize(updates, 2));
  3. df.write.format("org.apache.hudi").
  4.     options(getQuickstartWriteConfigs).
  5.     option(PRECOMBINE_FIELD_OPT_KEY, "ts").
  6.     option(RECORDKEY_FIELD_OPT_KEY, "uuid").
  7.     option(PARTITIONPATH_FIELD_OPT_KEY, "partitionpath").
  8.     option(TABLE_NAME, tableName).
  9.     mode(Append).
  10.     save(basePath);

注意，保存模式现在为追加。通常，除非您是第一次尝试创建数据集，否则请始终使用追加模式。 查询现在再次查询数据将显示更新的行程。每个写操作都会生成一个新的由时间戳表示的commit 。在之前提交的相同的_hoodie_record_key中寻找_hoodie_commit_time, rider, driver字段变更。

增量查询

Hudi还提供了获取给定提交时间戳以来已更改的记录流的功能。 这可以通过使用Hudi的增量视图并提供所需更改的开始时间来实现。 如果我们需要给定提交之后的所有更改(这是常见的情况)，则无需指定结束时间。

  1. // reload data
  2. spark.
  3.     read.
  4.     format("org.apache.hudi").
  5.     load(basePath + "/*/*/*/*").
  6.     createOrReplaceTempView("hudi_ro_table")
  8. val commits = spark.sql("select distinct(_hoodie_commit_time) as commitTime from  hudi_ro_table order by commitTime").map(k => k.getString(0)).take(50)
  9. val beginTime = commits(commits.length - 2) // commit time we are interested in
  11. // 增量查询数据
  12. val incViewDF = spark.
  13.     read.
  14.     format("org.apache.hudi").
  15.     option(VIEW_TYPE_OPT_KEY, VIEW_TYPE_INCREMENTAL_OPT_VAL).
  16.     option(BEGIN_INSTANTTIME_OPT_KEY, beginTime).
  17.     load(basePath);
  18. incViewDF.registerTempTable("hudi_incr_table")
  19. spark.sql("select `_hoodie_commit_time`, fare, begin_lon, begin_lat, ts from  hudi_incr_table where fare > 20.0").show()

这将提供在开始时间提交之后发生的所有更改，其中包含票价大于20.0的过滤器。关于此功能的独特之处在于，它现在使您可以在批量数据上创作流式管道。

特定时间点查询

让我们看一下如何查询特定时间的数据。可以通过将结束时间指向特定的提交时间，将开始时间指向”000”(表示最早的提交时间)来表示特定时间。

  1. val beginTime = "000" // Represents all commits > this time.
  2. val endTime = commits(commits.length - 2) // commit time we are interested in
  4. // 增量查询数据
  5. val incViewDF = spark.read.format("org.apache.hudi").
  6.     option(VIEW_TYPE_OPT_KEY, VIEW_TYPE_INCREMENTAL_OPT_VAL).
  7.     option(BEGIN_INSTANTTIME_OPT_KEY, beginTime).
  8.     option(END_INSTANTTIME_OPT_KEY, endTime).
  9.     load(basePath);
  10. incViewDF.registerTempTable("hudi_incr_table")
  11. spark.sql("select `_hoodie_commit_time`, fare, begin_lon, begin_lat, ts from  hudi_incr_table where fare > 20.0").show()

从这开始下一步？

您也可以通过自己构建hudi来快速开始， 并在spark-shell命令中使用--jars <path to hudi_code>/packaging/hudi-spark-bundle/target/hudi-spark-bundle-*.*.*-SNAPSHOT.jar， 而不是--packages org.apache.hudi:hudi-spark-bundle:0.5.0-incubating

这里我们使用Spark演示了Hudi的功能。但是，Hudi可以支持多种存储类型/视图，并且可以从Hive，Spark，Presto等查询引擎中查询Hudi数据集。 我们制作了一个基于Docker设置、所有依赖系统都在本地运行的演示视频， 我们建议您复制相同的设置然后按照这里的步骤自己运行这个演示。 另外，如果您正在寻找将现有数据迁移到Hudi的方法，请参考迁移指南。