知识中心

【译】Delta Lake 0.5.0介绍-阿里云开发者社区

2020-03-08 00:00:00 mimukeji

原文链接:https://databricks.com/blog/2020/01/29/query-delta-lake-tables-presto-athena-improved-operations-concurrency-merge-performance.html

最近,Delta Lake发布了0.5.0版本,该版本加入了对Presto和Athena的支持,以及提升了操作的并发性,本文将对Delta Lake 0.5.0版本的变化进行一个简单的介绍。

Delta Lake 0.5.0发布的几个最重要的特性如下:

  • 通过使用Manifest文件能够,支持其他数据处理引擎,现在能够使用Scala、Java、Python和SQL的API生成Manifest文件,并使用该文件通过Presto和Amazon Athena访问Delta Lake中的表数据,详细的使用方式见 https://docs.delta.io/0.5.0/presto-integration.html
  • 提升了Delta Lake所有操作的并发性,现在可以并发执行更多的Delta Lake操作。通过使用更加精细的冲突检测策略,Delta Lake的乐观并发控制得到了有效的改善,这使得我们能够更加容易地在Delta表上执行复杂的工作流。举例来说:

    1. 我们可以在给新分区添加内容的同时,在旧分区上执行delete操作;
    2. 在不相交的一组分区上同时执行update和merge操作;
    3. 让文件合并和增加文件内容同时执行

想要获取更多的信息,可以参考开源Delta Lake 0.5.0的release notes。在此博客文章中,我们将详细介绍如何使用Presto读取Delta Lake表、操作并发性的提升以及使用insert-only merge操作来更方便快速地去除重复数据。

使用Presto读取Delta Lake表

正如这篇文章Simple, Reliable Upserts and Deletes on Delta Lake Tables using Python APIs 所说,Delta Lake 的一些修改数据的操作,如delete操作,是通过给包含删除数据的文件写一个新版本,并只是将旧版本文件标记为已删除来实现的。Delta Lake采用这种方法的优势在于能够让我们查询旧版本的数据。如果我们想要了解哪些数据(或行)包含最新的数据,默认情况下我们可以去查询事务日志。其他数据处理系统,如Presto和Athena想要获取这些信息,可以通过读取Delta Lake生成的一种清单文本文件——Manifest,该文件中包含查询Delta Lake表需要读取的数据文件列表。为了实现Presto和Athena读取Delta Lake表,我们可以通过执行一些Python命令来实现,详细的内容可以参考 Set up the Presto or Athena to Delta Lake integration and query Delta tables

生成Delta Lake的Manifest文件

首先,使用以下代码片段创建Delta Lake的Manifest文件:

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

正如代码字面意思所示,以上操作将在表根目录中生成Manifest文件。如果你根据 Simple, Reliable Upserts and Deletes on Delta Lake Tables using Python APIs 这篇文章介绍的内容创建了departureDelays表,将会在表根目录中产生一个新的文件夹:

$/departureDelays.delta/_symlink_format_manifest

该文件夹中会有一个名为manifest的文本文件。如果你查看manifest文件的内容(例如使用cat命令),你将能看到类似以下的文本内容,它们指示了包含最新快照的文件。

file:$/departureDelays.delta/part-00003-...-c000.snappy.parquet
file:$/departureDelays.delta/part-00006-...-c000.snappy.parquet
file:$/departureDelays.delta/part-00001-...-c000.snappy.parquet
file:$/departureDelays.delta/part-00000-...-c000.snappy.parquet
file:$/departureDelays.delta/part-00000-...-c000.snappy.parquet
file:$/departureDelays.delta/part-00001-...-c000.snappy.parquet
file:$/departureDelays.delta/part-00002-...-c000.snappy.parquet
file:$/departureDelays.delta/part-00007-...-c000.snappy.parquet

创建Presto表以读取生成的Manifest文件

接下来的步骤是在Hive Metastore中创建一个外部表,以便Presto(或Athena)可以读取上一步生成的Manifest文件,来获得需要读取的Parquet文件,以读取Delta Lake表的最新快照。需要说明的是,对于Presto,你可以使用Apache Spark或Hive CLI来运行以下命令:

1. CREATE EXTERNAL TABLE departureDelaysExternal ( ... )
2. ROW FORMAT SERDE
   'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe'
3. STORED AS INPUTFORMAT
4. OUTPUTFORMAT
   'org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat'
5. LOCATION '$/departureDelays.delta/_symlink_format_manifest'

一些重要的说明:

  • 第一行所定义的schema必须和Delta Lake中的schema相同
  • 第五行需要指向Manifest文件的位置——_symlink_format_manifest

Presto(或Athena)需要配置SymlinkTextInputFormat才能从Manifest文件中获取Parquet数据文件的列表,而不是使用目录列表中的文件。需要说明的是,如果想要使用分区表,需要按照Configure Presto to read the generated manifests这篇文章进行一些额外的步骤。

更新Manifest文件

需要注意的是,如果Delata Lake的数据有更新,都需要重新生成Manifest文件,以便Presto能够获取到最新的数据。

操作并发性的提升

在Delta Lake 0.5.0版本,我们能够同时执行更多的操作。通过更细粒度的冲突检测,这些最新的更新让Delta Lake能够更容易地在Delta Lake表上执行复杂的工作流,例如:

  • 可以在给新分区添加内容的同时,在旧分区上执行delete操作;
  • 追加文件内容的同时执行文件合并操作;
  • 在不相交的一组分区上同时执行update和merge操作。

并发追加文件内容的用例

举个例子,当我们在执行merge操作的同时,如果有并发的事务向同一个分区写入记录,Delta Lake往往会抛出ConcurrentAppendException异常。

// Target 'deltaTable' is partitioned by date and country
deltaTable.as("t").merge(
    source.as("s"),
    "s.user_id = t.user_id AND s.date = t.date AND s.country = t.country")
  .whenMatched().updateAll()
  .whenNotMatched().insertAll()
  .execute()

上面的代码段就有可能会引发冲突,因为即使表已经按照date和country进行了分区,条件仍然不够明确。问题在于,这个查询将扫描整个表,从而可能与update任何其他分区的并发操作发生冲突。通过指定specificDatespecificCountry,以便可以在特定的date和country进行merge操作,现在我们就可以安全地在不同的date和contry同时执行此操作。

// Target 'deltaTable' is partitioned by date and country
deltaTable.as("t").merge(
    source.as("s"),
    "s.user_id = t.user_id AND d.date = '" + specificDate + "' AND d.country = '" + specificCountry + "'")
  .whenMatched().updateAll()
  .whenNotMatched().insertAll()
  .execute()

以上方法适用于其他所有的Delta Lake操作(如delete、更改元数据等)。

并发文件合并

如果你连续不断地将数据写入Delta表,随着时间的流逝,将会累积出大量的文件。这在流式数据场景中尤为重要,因为此时是以比较小的batch写入数据的,这将会导致文件系统不断地累积小文件,随着时间的推移,小文件的数量会不断增加,会降低查询的性能。优化这种场景的一个比较重要的方式就是定期获取大批量的小文件,并将其重写为数量比较小的大文件(文件合并)。
过去,在同时进行数据查询和执行文件合并时,出现异常的可能性会非常高。但是现在,由于Delta Lake 0.5.0版本的优化改进,我们可以同时执行查询操作(包括流式查询)和文件的合并,并且不会有任何异常产生。举个例子来说,如果你的表已经进行了分区,并且你只想基于谓词对一个分区进行重新分区,则可以使用where来仅读取该分区,并使用replaceWhere回写该分区:

path = "..."
partition = "year = '2019'"
numFilesPerPartition = 16   # Compact partition of a table to no. of files

(spark.read
  .format("delta")
  .load(path)
  .where(partition)
  .repartition(numFilesPerPartition)
  .write
  .option("dataChange", "false")
  .format("delta")
  .mode("overwrite")
  .option("replaceWhere", partition)
  .save(path))

以上代码中需要注意的是,仅在没有数据更改时,才使用dataChange == false选项,否则可能会破坏底层数据。

使用Insert-only Merge操作方便快速地去除重复数据

一个场景的ETL用例是搜集日志,并将其附加到Delta Lake表当中,一个比较常见的问题是数据源会产生重复的日志记录。通过使用Delta Lake的merge,你可以避免插入这些重复的记录,例如以下涉及merge以及update航班数据的代码:

# Merge merge_table with flights
deltaTable.alias("flights") 
    .merge(merge_table.alias("updates"),"flights.date = updates.date") 
    .whenMatchedUpdate(set = { "delay" : "updates.delay" } ) 
    .whenNotMatchedInsertAll() 
    .execute()

在Delta Lake 0.5.0版本之前,不可能从Delta Lake表中将重复数据作为流进行读取,因为insert-only merge并不是纯粹地将数据追加到表中。
例如,在流查询中,你可以在foreachBatch中执行merge操作来连续不断地将流数据写入Delta Lake表当中,并将需要删除的重复数据打上标记。以下PySpark的代码展示了这个场景:

from delta.tables import *

deltaTable = DeltaTable.forPath(spark, "/data/aggregates")

# Function to upsert microBatchOutputDF into Delta table using merge
def upsertToDelta(microBatchOutputDF, batchId):
  deltaTable.alias("t").merge(
      microBatchOutputDF.alias("s"),
      "s.key = t.key") 
    .whenMatchedUpdateAll() 
    .whenNotMatchedInsertAll() 
    .execute()
}

# Write the output of a streaming aggregation query into Delta table
streamingAggregatesDF.writeStream 
  .format("delta") 
  .foreachBatch(upsertToDelta) 
  .outputMode("update") 
  .start()

在另一个流式查询中,你可以从该Delta Lake表中连续读取需要删除的重复数据。这是可能的,因为insert-only merge操作(在Delta Lake 0.5.0版本引入)只会将新数据追加到Delta Lake表中。