⬜️Бизнес-задача:
— Операции с контрагентами совершаются в около 15 разных валют
— Есть необходимость пересчета финансовых показателей в разные валюты для отчетности
— Обменные курсы актуальны на каждую отдельную дату (курс меняется)
⬜️ Реализация:
— Используется поставщик данных и его API (для примера это Open Exchange Rates)
— Ежедневно несколько раз в течение дня (раз в 3 часа) делаются запросы на получение обменных курсов к каждой из 15 валют (Airflow)
— Ответы в виде JSON сохраняются AS IS в S3 / Object storage
— Эти файлы читаются на стороне DWH и формируется витрина обменных курсов (пар) в разрезе базовой валюты и даты
⬜️ Ключевые моменты, на которые стоит обращать внимание:
🟡Case sensitive identifiers
Трехбуквенных код валюты указывается в верхнем регистре. Учитывайте это при чтении данных, иначе получите NULLs вместо рейтов.
В случае Redshift + dbt я добавляю hooks:
pre_hook=["SET enable_case_sensitive_identifier TO TRUE"],
post_hook=["SET enable_case_sensitive_identifier TO FALSE"]
🟡Корректность файлов JSON содержащихся в S3
Порой ответы от API сервиса приходят с ошибкой, например:
<html>
<head><title>504 Gateway Time-out</title></head>
<body>
<center><h1>504 Gateway Time-out</h1></center>
</body>
</html>
Такие файлы, даже находясь среди корректных, могут вызывать ошибки чтения:
ERROR: Spectrum Scan Error
Detail:
-----------------------------------------------
error: Spectrum Scan Error
code: 15001
context: Error while reading next ION/JSON value: Invalid syntax. File: /dwh/currencies/2024-01-14-CHF-2024-01-14-08-15-10-UTC.json Offset: 0 Line: 1
🟡В рамках одних суток актуален последний курс
Т.е. из всех курсов в течение дня необходимо учитывать только тот, что приходится на закрытие дня.
Я делаю эту пометку и фильтр на нее с помощью window function:
SELECT
...
, (ROW_NUMBER() OVER (PARTITION BY base_currency, business_dt ORDER BY business_ts DESC) = 1) AS is_most_recent
...
WHERE ...
AND is_most_recent
🟡 Чтение огромного числа маленьких файлов
Требует гораздо большего количества времени нежели чем чтение одного (относительно) большого файла.
Поэтому, рекомендация:
— Делать compaction / merge файлов (сливать множество маленьких в один большой, например, раз в месяц)
— Выполнять загрузку в момент появления нового файла
С этим могут помочь инструменты типа EL SaaS (Hevo, Fivetran), или встроенные в СУБД возможности (Snowpipe для Snowflake)
Дайте знать, если полезно ⭐️
#s3 #object_storage #dwh #dbt #external_data
🌐 @data_apps | Навигация по каналу