Как использовать DBT с AWS Athena с таблицами Apache Iceberg
У нас есть модели DBT, которые мы используем для запуска на столах AWS Athena. Он создает внешние таблицы Hive за кулисами. Теперь у нас есть ситуация, когда тип данных столбца может измениться в будущем. Таблицы Athena на основе Hive не позволяют изменить тип данных столбца, но таблица Apache Iceberg позволяет. Мы можем изменить тип данных столбца в таблицах Apache Iceberg.
Мы скопировали данные из старой таблицы Hive в таблицу Iceberg, но когда мы запускаем модель DBT, появляется следующая ошибка:
[error] [MainThread]: An error occurred (InvalidInputException) when calling the GetPartitions operation:
Конфигурация DBT для модели выглядит следующим образом. Раньше он работал с внешними таблицами Hive, но не работал с таблицами Apache Iceberg.
{{
config(materialized='incremental',
external_location = "s3://" + env_var('BUCKET-NAME') + "/" + env_var('SCHEMA-NAME') + "/" + this.identifier,
partitioned_by = ['event_date'],
incremental_strategy='insert_overwrite',
on_schema_change='ignore'
)
}}
Apache Iceberg создается следующим образом:
CREATE TABLE iceberg_table (
id int,
data string,
event_date string)
PARTITIONED BY (event_date)
LOCATION 's3://DOC-EXAMPLE-BUCKET/iceberg-folder'
TBLPROPERTIES (
'table_type'='ICEBERG',
'format'='parquet',
'write_target_data_file_size_bytes'='536870912',
'optimize_rewrite_delete_file_threshold'='10'
)
Ответы 2
Я узнал, что на данный момент DBT не поддерживает Apache Iceberg. Вместо этого мы должны использовать Apache Hudi.
DBT с Athena поддерживает Apache Iceberg. Кусок кода вашей конфигурации будет выглядеть так:
{{
config(
schema = env_var('DATABASE'),
s3_data_dir='s3://' ~ env_var('BUCKET') ~ '/',
s3_data_naming='table_unique',
format='parquet',
write_compression='GZIP',
materialized='incremental',
table_type='iceberg',
incremental_strategy = 'merge',
unique_key = ['key1', 'key2],
tags=["insert_tags"]
)
}}
Затем запуск вашей модели DBT должен создать таблицу Iceberg. В этом случае он будет следовать стратегии upsert, обновляя значения с изменениями или вставляя новые на основе поля unique_key.