Delta Lake Presto Integration & Manifests 机制-阿里云开发者社区

原文链接

该功能与我们之前平台化 Delta Lake 平台化实践（离线篇) 的很多工作都较为相似，比如与 metastore 的集成，直接通过 manifest 读取 delta 存活文件等。
Delta Lake 在 0.5 之前只支持通过 Spark 读取数据，在新版本中增加了其他处理引擎通过 manifest 文件访问 Delta Lake 的能力。下文以Presto 为例说明如何通过 manifest 文件访问数据，manifest 文件的生成及其一些限制。

01 使用

Presto 使用 manifest 文件从 hive 外部表中读取数据，manifest文件是一个文本文件，包含该表/分区所有存活数据的路径列表。

当使用 manifest 文件在 Hive metastore 中定义外部表时，Presto 将会先读取 manifest 中的文件路径列表再去访问想要的文件，而不是直接通过目录列表来查找文件。

1.1 通过 Spark 生成 manifest 文件

支持 sql / scala / java / python 四种 api，以 sql 和 scala 为例。

SQL

GENERATE symlink_format_manifest FOR TABLE delta.`pathToDeltaTable`

Scala

val deltaTable = DeltaTable.forPath(pathToDeltaTable)
deltaTable.generate("symlink_format_manifest")

使用 GENERATE 命令会在/path/to/deltaTable/_symlink_format_manifest/ 目录下生成一个 manifest 文件，其中包含了所有存活的文件路径。

查看清单文件

cat /path/to/deltaTable/_symlink_format_manifest/manifest

hdfs://tdhdfs-cs-hz/user/hive/warehouse/bigdata.db/delta_lsw_test/part-00000-0a69ce8d-0d9e-47e2-95b2-001bd196441d-c000.snappy.parquet
hdfs://tdhdfs-cs-hz/user/hive/warehouse/bigdata.db/delta_lsw_test/part-00000-ba1767cb-ff0e-4e65-8e83-7a0cdce6a2f4-c000.snappy.parquet

如果是分区表，例如以 ds 作为分区字段，生成的结构如果下，每个分区下都有一个 manifest 文件包含了该分区的存活文件路径。

/path/to/table/_delta_log
/path/to/table/ds=20190101
/path/to/table/ds=20190102
/path/to/table/_symlink_format_manifest
---- /path/to/table/_symlink_format_manifest/ds=20190101/manifest
---- /path/to/table/_symlink_format_manifest/ds=20190102/manifest

存活文件定义：add file - remove file

1.2 定义 Hive Metastore 外部表读取相应文件

CREATE EXTERNAL TABLE mytable ( ... )   -- 与 Delta table 一致的 schema 信息
PARTITIONED BY ( ... )  -- 分区参数可选，需要与 Delta table 一致
ROW FORMAT SERDE 'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe'
STORED AS INPUTFORMAT 'org.apache.hadoop.hive.ql.io.SymlinkTextInputFormat'
OUTPUTFORMAT 'org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat'
LOCATION '<pathToDeltaTable>/_symlink_format_manifest/'  -- 指定 manifest 地址

通过 SymlinkTextInputFormat ，Presto 可以直接从 manifest 中读取需要的文件而不需要直接定位到数据目录。

如果是分区表的话，在运行 generate 后，需要运行 MSCK REPAIR TABLE 使 Hive Metastore 能发现最新的分区。使用 repair 有两种场景：

每次清单文件生成后调用：每次 generate 都调用 repair，这种方式在分区多的情况下性能表现会非常糟糕，我们的做法是在数据写入时从 spark 获取到相应的变更分区然后依次执行 ADD PARTITION操作。
在需要新分区的时候调用：如果是按天粒度的分区表，可以选择在午夜12点创建新分区同时执行 generate 后运行一次 repair 。

important: 如果使用了 kerberos 认证，必须要在 presto 目录的etc/catalog/hive.properties 中配置 yarn-site.xml，否则在查询数据时会提示错误

com.facebook.presto.spi.PrestoException: Can't get Master Kerberos principal for use as renewer
  at com.facebook.presto.hive.BackgroundHiveSplitLoader$HiveSplitLoaderTask.process(BackgroundHiveSplitLoader.java:191)
  at com.facebook.presto.hive.util.ResumableTasks.safeProcessTask(ResumableTasks.java:47)
  at com.facebook.presto.hive.util.ResumableTasks.access$000(ResumableTasks.java:20)
  at com.facebook.presto.hive.util.ResumableTasks$1.run(ResumableTasks.java:35)
  at io.airlift.concurrent.BoundedExecutor.drainQueue(BoundedExecutor.java:78)
  at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
  at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
  at java.lang.Thread.run(Thread.java:748)
Caused by: java.io.IOException: Can't get Master Kerberos principal for use as renewer
  at org.apache.hadoop.mapreduce.security.TokenCache.obtainTokensForNamenodesInternal(TokenCache.java:116)
  at org.apache.hadoop.mapreduce.security.TokenCache.obtainTokensForNamenodesInternal(TokenCache.java:100)
  at org.apache.hadoop.mapreduce.security.TokenCache.obtainTokensForNamenodes(TokenCache.java:80)
  at org.apache.hadoop.mapred.FileInputFormat.listStatus(FileInputFormat.java:206)
  at org.apache.hadoop.mapred.FileInputFormat.getSplits(FileInputFormat.java:315)
  at com.facebook.presto.hive.BackgroundHiveSplitLoader.loadPartition(BackgroundHiveSplitLoader.java:304)
  at com.facebook.presto.hive.BackgroundHiveSplitLoader.loadSplits(BackgroundHiveSplitLoader.java:258)
  at com.facebook.presto.hive.BackgroundHiveSplitLoader.access$300(BackgroundHiveSplitLoader.java:93)
  at com.facebook.presto.hive.BackgroundHiveSplitLoader$HiveSplitLoaderTask.process(BackgroundHiveSplitLoader.java:187)
  ... 7 more

02 Generate 过程

Generate 命令生成 manifest 的逻辑并不复杂，有兴趣的同学可以看下，方法入口：

DeltaGenerateCommand ->

GenerateSymlinkManifest.generateFullManifest(spark: SparkSession,deltaLog: DeltaLog)

在分区表每个分区或者非分区表中原子性的更新 manifest 文件

def writeSingleManifestFile(
manifestDirAbsPath: String,
dataFileRelativePaths: Iterator[String]): Unit = {
val manifestFilePath = new Path(manifestDirAbsPath, "manifest")
      val fs = manifestFilePath.getFileSystem(hadoopConf.value)
      fs.mkdirs(manifestFilePath.getParent())

      val manifestContent = dataFileRelativePaths.map { relativePath =>
        DeltaFileOperations.absolutePath(tableAbsPathForManifest, relativePath).toString
      }
      val logStore = LogStore(SparkEnv.get.conf, hadoopConf.value)
      logStore.write(manifestFilePath, manifestContent, overwrite = true)
    }

    // 我修复了 Delta 0.5 删除非分区表失效的 BUG，已将 PR 提交社区
    val newManifestPartitionRelativePaths =
      if (fileNamesForManifest.isEmpty && partitionCols.isEmpty) {
        writeSingleManifestFile(manifestRootDirPath, Iterator())
        Set.empty[String]
      } else {
        withRelativePartitionDir(spark, partitionCols, fileNamesForManifest)
          .select("relativePartitionDir", "path").as[(String, String)]
          .groupByKey(_._1).mapGroups {
          (relativePartitionDir: String, relativeDataFilePath: Iterator[(String, String)]) =>
            val manifestPartitionDirAbsPath = {
              if (relativePartitionDir == null || relativePartitionDir.isEmpty) manifestRootDirPath
              else new Path(manifestRootDirPath, relativePartitionDir).toString
            }
            writeSingleManifestFile(manifestPartitionDirAbsPath, relativeDataFilePath.map(_._2))
            relativePartitionDir
        }.collect().toSet
      }

删除分区表中失效分区的 manifest 文件

  val existingManifestPartitionRelativePaths = {
      val manifestRootDirAbsPath = fs.makeQualified(new Path(manifestRootDirPath))
      if (fs.exists(manifestRootDirAbsPath)) {
        val index = new InMemoryFileIndex(spark, Seq(manifestRootDirAbsPath), Map.empty, None)
        val prefixToStrip = manifestRootDirAbsPath.toUri.getPath
        index.inputFiles.map { p =>
          val relativeManifestFilePath =
            new Path(p).toUri.getPath.stripPrefix(prefixToStrip).stripPrefix(Path.SEPARATOR)
          new Path(relativeManifestFilePath).getParent.toString 
        }.filterNot(_.trim.isEmpty).toSet
      } else Set.empty[String]
    }

    val manifestFilePartitionsToDelete =
      existingManifestPartitionRelativePaths.diff(newManifestPartitionRelativePaths)
    deleteManifestFiles(manifestRootDirPath, manifestFilePartitionsToDelete, hadoopConf)

03 一些限制

3.1 数据一致性

在 Delta Lake 更新 manifest 时，它会原子的自动覆盖现有的 manifest 文件。因此，Presto 将始终看到一致的数据文件视图，然而，保证一致性的粒度取决于表是否分区。

非分区表

所有的文件路径都写在一个会原子更新的 manifest 文件中（参考上文结构），这种情况下 Presto 能看到一致性快照。

分区表

manifest 文件将以 hive 分区的目录结构（参考上文结构），这意味着每个分区都是原子更新，所以 Presto 能看到一个分区内的一致性视图而不是跨分区的一致性视图。此外，由于所有的分区并不是同时更新，所以读取时可能会在不同分区中读到不同 manifest 版本。

简单的说，当在更新清单文件时，Presto 发起读请求，由于 manifest 所有分区并不是一次原子更新操作，所以有可能得到的结果并不是最新的数据。

3.2 性能

大量的文件数量会造成 Presto 性能下降，官方的建议是在执行 generate 生成 manifest 前先对文件进行 compact 操作。分区表的单个分区或是非分区表的文件数量不超过1000。

3.3 Schema 推断

原生的 Delta Lake 支持 schema evolution，意味着无论 hive metastore 定义的 schema 如何，都会基于文件使用最新的 Schema。由于 Presto 直接使用了定义在 hive metastore 中的 schema ，所以如果要修改 schema 信息，必须要对表进行相应更新。

04 后记

一些BUG

测试过程中还发现了一个 BUG，如果将非分区表的数据全部删除，则 generate 后 manifest 不会更新。
已将 PR 提交社区 https://github.com/delta-io/delta/issues/275

实践经验

首先，由于需要额外的调用 generate 命令生成/更新 manifest 文件，使用体验肯定不如直接通过 Spark 读取数据。
其次，在 generate 过程中进行数据读取有可能会遇到跨分区查询版本不一致的情况，但是瑕不掩瑜，通过 manifest，与大数据生态其他处理引擎的道路被打开了。
就像在 Delta Lake 平台化实践（离线篇) 这篇文章中提到的，我们的大数据平台有一个功能是表数据/分区数据预览，通过 spark 去查用户体验会相当差（耗时长），我们之前的做法是自定义了一个工具类在查询时从 _delta_log中生成 manifest，再通过 manifest 获取到的文件路径直接从文件系统中读取 Parquet 实现，有了 generate 功能，就可以直接读取 manifest 文件，外部系统扩展工作量极大的简化。
在我们的生产环境中，presto 和 spark 使用的同一套 hive metastore ，但是 spark 直接读取上述创建的外部表会报错（就算能读也会有一致性风险），解决办法是在平台拦截了 sql 方法，通过判断 table properties 识别 delta 表，然后将其直接转化为 delta api 对数据进行操作，Presto 则是直接访问外表，解决了冲突的问题。

----

阿里巴巴开源大数据技术团队成立Apache Spark中国技术社区，定期推送精彩案例，技术专家直播，问答区近万人Spark技术同学在线提问答疑，只为营造纯粹的Spark氛围，欢迎钉钉扫码加入！

对开源大数据和感兴趣的同学可以加小编微信（下图二维码，备注“进群”）进入技术交流微信群。

Apache Spark技术交流社区公众号，微信扫一扫关注

Delta Lake Presto Integration & Manifests 机制