databricks - 如何使用 COPY INTO 命令？

Question

我目前正在尝试实现一些 delta Lake 表来存储我们的“基础”数据。

这些表将根据以以下格式摄取到我们数据湖中“原始”区域的增量数据构建：

• /raw/MyDataSource/2020/11/23/EntityOne_1056.parquet
• /raw/MyDataSource/2020/11/23/EntityTwo_1059.parquet
• /raw/MyDataSource/2020/11/22/EntityOne_0956.parquet

我可以使用一系列循环和数据框来读取现有数据并将其与新数据合并，但是我认为有更简单的方法吗？

我发现了COPY INTO命令（https://docs.databricks.com/spark/latest/spark-sql/language-manual/delta-copy-into.html），它似乎让合并和管理哪个批处理文件已被处理到 delta 湖表中，但我不确定如何实现这一点，我能找到的唯一示例非常有限。

  COPY INTO table_identifier
  FROM [ file_location | (SELECT identifier_list FROM file_location) ]
  FILEFORMAT = data_source
  [FILES = [file_name, ... | PATTERN = 'regex_pattern']
  [FORMAT_OPTIONS ('data_source_reader_option' = 'value', ...)]
  [COPY_OPTIONS 'force' = ('false'|'true')]

我想我首先需要创建一个预期实体名称的列表，然后使用它COPY INTO为每个实体/增量表运行一个命令。

我需要让命令查看数据湖中的“原始”区域并找到该实体的所有文件（可能在文件路径中使用通配符或 RegEx 模式）。就像是：

"/raw/MyDataSource/*/*/*/{0}_*.parquet".format("EntityOne")

最后，我需要使用spark.sql(..)命令或使用我的 Databricks 笔记本中的 SQL 单元来调用命令查询。

我的问题：

1. 我是否了解该COPY INTO命令可以正确执行的操作？
2. 如何COPY INTO使用我的用例构建查询？

标签： databricksdelta-lake