Data Engineering on AWS Learning Plans

Это статья краткая заметка по курсу Data Engineering on AWS Learning Plans (includes labs)
В некотором роде - это аналог курса Инженер облачных сервисов Yandex, но, по ощущениям, имеет меньшую техническую составляющую.

• Foundations
• Сервисы для формирования облака
• Доступ в сеть
• Сost optimization
• Права доступа
• Data Lake Solution
• S3
• Data copy
• Glue
• Трансформация и визуализация
• Security
• DWH
• Redshift
• Redshift monitoring
• Redshift schedulling
• Redshift права доступа
• EMR
• EMR optimization
• EMR права доступа
• Stream processing
• Kinesis
• Firehose
• Integration

Foundations

Сервисы для формирования облака

- AWS CloudFormation
JSON or YAML file that defines the AWS infrastructure resources
Opensource аналог - Terraform.
- CDK - Cloud Development Kit
Higher-level, more developer-friendly abstraction on top of CloudFormation
Uses lang as py or js.
- SAM - Serverless Application Model
Еще абстракция над CloudFormation с использованием шаблонов AWS SAM template specification
AWS SAM CLI - подготовка и запуск шаблонов.
Часто используется для запуска aws lambda functions

Доступ в сеть

- AWS Site-to-Site VPN - коннект между локальной сетью и облаком
- AWS Direct Connect - прямой коннект, минуя публичные сети
- AWS PrivateLink - может быть как AWS Direct Connect + прямой коннект между чужим AWS аккаунтом
- VPC endpoints - эндпойны для связи сервисов между собой, без выхода в инет

Сost optimization

Основные возможности:
Динамический размер группы машин
- AWS Auto Scaling - auto scale your VM groups
Резервирование инстансов на более 1 года
Спот инстансы - с возможностью выключения
Тегирование всех ресурсов и их учет:
- AWS Cost Explorer - costs and usage patterns - AWS Budgets - alert on overspendings
- AWS Cost Anomaly Detection - алерты о нетипичной нагрузке
Экономия на хранении
- AWS data transfer pricing tier - reduce costs for move data from diff regions
- Amazon S3 Intelligent-Tiering - автосмена типа хранения с3 на более дешевый

Права доступа

IAM policies - json с правами
Identity-based policies - права и роли пользователя или группы
Resource-based policies - разрешенные действия над ресурсом
- по ролям
- предустановленные полиси
- по установленному тегу
- атрибуты типа: время, расположение, девайс и прочее
IAM groups - группа пользователей
IAM roles - права, которые даются на сервис или инстанс
IAM endpoints - на что даем права

AWS KMS - хранилище секретов (аудит, права, доступ, ротация), ключи шифрования

Data Lake Solution

S3

Типы хранения:
- Standart
- S3 Intelligent-Tiering - автоматическое определение уровня хранения
- S3 Express One Zone - OLTP (single zone)
- S3 Standard-Infrequent Access (S3 Standard-IA) - быстрый но редкий доступ
- S3 One Zone-Infrequent Access (S3 One Zone-IA) - быстрый, редкий, 1 зона
- S3 Glacier Instant Retrieval - еще реже (раз в год), но быстрый доступ
- S3 Glacier Flexible Retrieval - бэкапы, с большой пропускной способностью, но подготовка от 1 минуты
- S3 Glacier Deep Archive - холоднющее

S3 Storage Class Analysis - выявление частоты использования
S3 Storage Lens - автоматические инсайты
используя это настраиваем S3 Lifecycle policies (не забываем про версии, про битые мультипарт загрузки, так же зависит от числа объектов)

Data copy

- DMS / Database Migration Service - copy data db-db, db-s3
Нужен репликейшен инстанс для запуска
* full или cdc + инфа о конекте from/to
(после фулл есть этап донаката накопленных данных во время первичной заугрузки)
так же автоматически заполняется glue catalog с информацией о таблицах
Opensource аналог - Debezium + Kafka Streams.

- DataSync - для копирования файлов

Glue

- Glue = hive metastore + crowler + возможности кверить через spark, athena, redshift
автоматически парсит папку на базу (поддерживает таблицы и партиции) по расписанию (умеет в доскан)
интегрируется с Amazon Athena / EMR
хранится как в хайв каталоге: таблица, тип, путь, партиции и пути, параметры, инфер схемы (колонки и типы), статистика на таблице и колонке
только 1 каталог в регионе
есть история изменения схемы
может работать не только с s3, но и другими авс базами через jdbc + kafka
* AWS Glue Studio (visual ETL, notebooks, script editor) - Python or Spark under the hood
Databrew - visual ETL for glue

Трансформация и визуализация

- EMR - только Spark в servless иначе полные набор: Hadoop+Spark+Hive
EMR studio = jupiter
* автодетект сенсивной информации (свои правила и регексп)
* дата кволити - описывается правилами (может фильтроваться/отправляться)
* job monitoring - потребление ресурсов, анализ ресурсов и логов
- Athena - servless, basic transformations = Trino, but pay per usage
- Lambda - servless функции (по расписанию или событию)

- AWS Step Functions / Airflow - orchestration

- QuickSight - like powerbi (есть inmem calc)
direct запросы или полная загрузка в inmem (SPICE) и использование внутреннего движка

Security

- IAM - основа всех правил
- AWS Lake Formation = Ranger (взаимодействует с каталогом) + утилиты ingestion
доступы могут быть на ресурс и на тег (тег можно задать для группы таблиц или колонок)
доступы можно выдавать и на s3
trusted aws сервисы получают временные права, после проверки доступов в Lake Formation
- в redshift / athena - права уже выдаются внутри (grant)
- emr (spark) / glue - выдается через iam роли
- CloudTrail - аудит доступов и вызовов

DWH

Redshift

Redshift - MPP db, сама скалируется, интегрируется с с3 и спарк, читает паркет/айсберг, есть бэкапы/кодирование, федеративные запросы
есть поддержка единичных (Oltp) запросов

Режимы работы:
- развертывание на своих виртуалках
архитектура:
-- Redshift Managed Storage (RMS) - специальный тир хранения данных на s3 (может скейлится отдельно от компьюта)
-- ведущая нода делить данные на слайсы согласно размера ноды и данных, планирует запрос на части
обеспечивает jdbc pg лайк драйвера
-- ведомые компьют ноды - хранят данные. Расчет запускается, только если данные таблицы есть на этой ноде (shared nothing)
- серверлесс - сам скейлит ресурсы (можно задать лимит)
данные хранятся в отдельном слое, отдельно от компьют нод
платится только за использование

Workgroups - позволяют изолировать ресурсы пользователей (namespace = schemas)

Загрузка данных:
- copy from s3

copy T from 's3://path' authorization;

обратное направление:

unload ('select * from venue')
to 's3://mybucket/tickit/unload/venue_'
iam_role 'arn:aws:iam::0123456789012:role/MyRedshiftRole';

- AWS DMS (CDC) - zero etl (репликация данных)
- emr (для предагрегации и записи)
- glue - для каталогизации и обработки
- Amazon Redshift Data API - работа без jdbc через асинхронное api
- потоковая загрузка
-- можно стримить из кафки напрямую в matview
но ее нужно обновлять по расписанию (микробатч)
-- firehorse - микробатч раз в 60 секунд (загружается командой copy)
-- Lambda / emr - для батч загрузки

- udf - python (deprecated) OR any language using aws lambda function OR pg/sql (transactional. Есть ограничения на фунации)
- Redshift Spectrum - возможность читать данные напрямую с с3 без загрузки внутрь (но это отдельная виртуалка)
через нее обеспечивается пуш предикатов
- есть веб интерфейс для запросов и визуализации результатов
- есть matview с предрасчетов
есть автореврайт запросов, автоапдейт при изменении источника и автоматическое определение инкремента
AutoMV - рекомендации по созданию новых MV
- есть встроенные механизмы ML
- шаринг: именновая папка с таблицами, схемами и т.д. только для чтения (шарится между редшифт серверами)

Redshift monitoring

- CloudWatch - графики с метриками нагрузки + Логи
- Redshift Advisor - рекомендации по изменению конфигурации + алерты о выходе за границы
sort, distribution, vacuum, files compaction, unused resources, тип сжатия для колонки
данные так же лежат в sys* вьюхах (SVL_QUERY_REPORT = ASH)
есть метрики на уровне кластера, ноды и очереди
distribution: round robin, by key, all nodes
pk/fk есть только в схеме, но могут помочь планировщику ресурсов
* STL_QUERY - список запросов

Redshift schedulling

- Step Functions - типа аирфлоу, но с глубокой интеграцией с aws (есть и мэнэджет аирфлоу)
Workflow Studio - визуальный редактор dag
есть параллелизация, ожидание, мап
можно делать шаблоны и переиспользовать их
есть возможность запуска расчета по событию из потока

Redshift права доступа

- Авторизация: права на управление кластером, сетевой доступ до, права до объектов бд через IAM роли (best pract) или через grant внутри,
временные учетки только на время коннекта (федеративные пользователи из ad)
- есть RLS, CLS, шифрование, маскирование

---CLS
GRANT SELECT (c_custkey, c_acctbal, c_mktsegment, c_nationkey, c_comment) ON sales.customer 
TO ROLE sales_ro;

---RLS
CREATE RLS POLICY data_analyst_policy
WITH (c_nationkey bigint) AS cs
USING (
cs.c_nationkey IN (select n_nationkey from sales.nation n join sales.employee e on trim(n.n_name) = e.country_name and employee_user_name = current_user and e.employee_status='A'));

-- Grant select on Lookup tables
GRANT SELECT ON sales.nation ,sales.employee
TO RLS POLICY data_analyst_policy; 

-- Attach policy on the table to role
ATTACH RLS POLICY data_analyst_policy
ON sales.customer
TO ROLE sales_ro;

--Enable RLS security on tables
ALTER TABLE sales.customer ROW LEVEL SECURITY on;

Masking

--create a masking policy that fully masks the credit card number
CREATE MASKING POLICY mask_credit_card_full
WITH (credit_card VARCHAR(256))
USING ('000000XXXX0000'::TEXT);

ATTACH MASKING POLICY mask_credit_card_full
ON credit_cards(credit_card)
TO PUBLIC;

- есть аудит действий пользователя - CloudTrail -> CloudWatch (по умолчанию выключено)

EMR

- EMR == YA Dataproc
EMRFS - выглядит как hdfs, но на самом деле s3 (данные сохраняются при пересоздании)
- EventBridge - обработка событий из разных aws сервисов (like af sensors)
-- cron events,
-- по событию из Amazon S3, Amazon DynamoDB, Amazon Kinesis
s3 events notifications создаются прямо в настройках бакета - запускать lambda function (которая уже запустит степ) -- любые другие aws события

EMR optimization

- EMR:
configuration, spot instances (до 90% дешевле), auto scaling, servless?, partioning, monitoring queries
- Glue
Job bookmarks - остановка и продолжение с определенной точки (для инкрементальной загрузки и процессинга)
партицирование, мониторинг, шедулинг,
flex jobs - джобы для работы с метаданными (отдельно от основного компьюта)
так же есть встроенный шедулятор по типу airflow

EMR права доступа

- EMR: IAM, RANGER, VPC, KERBEROS, ENCRUPTION
- GLUE - +шифрование метаданных, KMS для хранения паролей и шифрование пользовательских данных
- Simple Notification Service - отправка всяких уведомлений

Stream processing

Kinesis

Kinesis - Kafka
- может запускать Lambda functions по событию прихода данных в kinessis (синхронный вызов)
- чекпойнт для промежуточного сохранения системы
- до 10 лямбд на партицию
- сообщения

MSK - managed Kafka

Firehose

- как коннекторы (sync) в kafka streams
читает в 2 режимах:
- опрос раз в 200 мс
- kinessis пушит сообщение незамедлительно (платный режим) - 70ms
обработка: смена формата данных, шифрование, может вызывать lambda

Integration

- Glue - поддерживает real time jobs (Apache Spark Structured Streaming или Ray фреймворк)
- Open search - можно слкдадывать для последующего полнотекстового анализа
- Redshift - через матвью
- DynamoDB - для быстрой вставки стрим данных
- Athena - тоже может читать данные из Kinesis напрямую
- AWS Flink - для near real time процессинга (но нет ingest)
пример создания клик стрима в флинке:

DROP TABLE IF EXISTS clickstream_events;
CREATE TABLE clickstream_events (
    event_id STRING,
    event STRING,
    user_id STRING,
    item_id STRING,
    item_quantity BIGINT,
    event_time TIMESTAMP(3),
    os STRING,
    page STRING,
    url STRING
  )
WITH (
    'connector' = 'kinesis',
    'stream' = '****',
    'aws.region' = 'us-east-1',
    'scan.stream.initpos' = 'LATEST',
    'format' = 'json'
);

join 2 клик стримов:

SELECT  *
from clickstream_events 
inner join  catalog_items_stream
on   clickstream_events.item_id = catalog_items_stream.item_id;

аутпут можно сохранять в стрим топик

DROP TABLE IF EXISTS sink_table;
CREATE TABLE sink_table (
    event_id STRING,
    event STRING,
    user_id STRING,
    item_id STRING,
    item_quantity BIGINT,
    event_time TIMESTAMP(3),
    os STRING,
    page STRING,
    url STRING,
    item_name STRING,
    item_price STRING
)
WITH (
'connector' = 'kinesis',
'stream' = '****',
'aws.region' = 'us-east-1',
'scan.stream.initpos' = 'LATEST',
'sink.producer.aggregation-enabled' ='false',
'format' = 'json'
);

и обращаться к нему как к таблице:

INSERT INTO sink_table
SELECT  
 event_id ,
    event ,
    user_id,
    catalog_items_stream.item_id,
    item_quantity,
    event_time,
    os,
    catalog_items_stream.page,
    url,
    item_name,
    item_price
from clickstream_events 
inner join catalog_items_stream
on   clickstream_events.item_id = catalog_items_stream.item_id;

SELECT * FROM sink_table;

- все эти таблицы автоматически регистрируются в glue каталоге !
- анализ окном в 10 секунд

 SELECT
        TUMBLE_START(PROCTIME(), INTERVAL '10' seconds) as start_window,
        TUMBLE_END(PROCTIME(), INTERVAL '10' seconds) as end_window,
        clickstream_events.page,  
        SUM(CAST(item_price as FLOAT) * item_quantity) AS SALES
from clickstream_events 
inner join catalog_items_stream
on   clickstream_events.item_id = catalog_items_stream.item_id
WHERE (event= 'purchased_item')
GROUP BY TUMBLE(PROCTIME(), INTERVAL '10' seconds ),clickstream_events.page, item_price;