Análisis de datos - DA SQL

Основное про Apache Airflow Apache Airflow - платформа с довольно обширным функционалом, но в контексте аналитики данных используется по большей части как ETL-инструмент, а также как инструмент для автоматизации процессов через питон. По сути, основная часть Airflow (особенно в дата-аналитике) - это python-код, который имеет некоторые различия с обычным питоном. Так как мы имеем дело с фреймворком, то мы можем пользоваться особыми классами (по типу DAG, BashOperator и т.п.). Также нужно учитывать, что при работе с Airflow мы не пишем то, как это должно выполняться. Тут используется более декларативный подход. Среди этого "питоновского диалекта" главное, на что следует обратить внимание - это DAG. DAG (Directed Acyclic Graph) - это рабочий процесс, который состоит из задач и их зависимостей. В нем задачи выполняются в строгом порядке, а также нельзя зациклить выполнение этих задач. Основа, для чего существует DAG, выглядят так: ▶️Описать сами задачи, в чем они заключаются ▶️Задать порядок выполнения этих задач ▶️Задать периодичность выполнения всего этого процесса Сам DAG создается так (пример): with DAG( # Базовые параметры dag_id='my_sample_dag', default_args=default_args, description='Пример DAG', start_date=datetime(2024, 1, 1), # Расписание schedule_interval='@daily', # или cron-строка '0 3 * * *' # Дополнительные настройки catchup=False, tags=['example', 'tutorial'], max_active_runs=1 ) as dag: (перед этим можно задать аргументы, которые передадутся в DAG, например:) default_args = { 'owner': 'admin', 'email': '[email protected]' } После этого пишутся задачи, в основном с использованием какого-либо оператора. Как пример: process_data = PythonOperator( task_id='process_data', # Вызов python-функции python_callable=python_function, # передача контекста в функцию provide_context=True ) и определяется зависимость между задачами start_task >> process_data >> end_task Сами вышеупомянутые операторы бывают: ▶️BashOperator - для Bash-команд ▶️PythonOperator - для питона ▶️PostgresOperator - для Postgres ▶️DummyOperator - действий не делает, только "группирует" таски ▶️MySqlOperator - для MySQL ▶️BranchPythonOperator - для сложных Python-кодов с условиями (если... то...) ▶️BranchSQLOperator - для сложных SQL-запросов с условиями (если... то...) ▶️FileSensor - проверяет наличие файла в директории и много других. 🟠В заключение как может в общих чертах выглядеть один из ETL-процессов в Airflow: from airflow import DAG from airflow.operators.python import PythonOperator from datetime import datetime def extract(): ... def transform(): ... def load(): ... with DAG( "my_etl_dag", start_date=datetime(2025, 1, 1), schedule_interval="@daily", ) as dag: extract_task = PythonOperator(task_id="extract", python_callable=extract) transform_task = PythonOperator(task_id="transform", python_callable=transform) load_task = PythonOperator(task_id="load", python_callable=load) extract_task >> transform_task >> load_task Сперва как обычно импорт библиотек и написание типичных питоновских функций (в данном случае по ETL). Потом создаем DAG, который будет выполняться ежедневно, начиная с 01.01.25. Далее через PythonOperator задаем логику для каждого из 3 ETL тасков, а именно: задаем id и передаем в таски ранее написанные функции. В самом конце строим порядок для выполнения этих 3 задач.

Обработка больших данных в Python - Dask Dask - библиотека в Python, которая позволяет работать с большими данными, когда обычные инструменты (Pandas, NumPy) тормозят или когда они не могут обработать большой объем данных из-за ограничений оперативной памяти. Dask разбивает большие массивы/датафреймы на части, которые обрабатываются параллельно. Также совместима с Pandas, NumPy, Scikit-learn и т.п. Целесообразно использовать такую библиотеку, если объем данных превышает 10 ГБ (порог может быть ниже и выше в зависимости от RAM). 🟠Установка через pip: pip install dask 🟠Основные функции: ▶️Dask Arrays import dask.array as da x = da.random.random((100000, 100000), chunks=(1000, 1000)) y = x.mean().compute() Здесь был создан большой массив из случайных чисел (100 000 х 100 000), который был разбит на блоки (1000 х 1000) и по которому было посчитано среднее значение. Вывести его через print не удастся (в отличие, например, от меньшего массива в Numpy) из-за ограничений памяти, поэтому с ним можно работать только по части метаинформации, либо по части результатов вычислений. ▶️Dask DataFrames import dask.dataframe as dd df = dd.read_csv(r"ПУТЬ\file.csv", encoding="utf-8", sep=';') result = df.groupby('col_name').sum().compute() print (result) Подключение к большому CSV-файлу (по аналогии с Pandas) и группировка значений (по принципу group by в sql). Но по большому счету что массивы, что CSV-файлы на несколько десятков ГБ особо не встречаются. На мой взгляд, всю свою полезность Dask раскрывает в связке с базами данных (к которым можно подключиться, например, через cx_oracle) 🟠Вообще, функционал Dask в целом напоминает функционал Pandas и Numpy. Но у него есть 2 ключевые особенности, связанные именно с его спецификой: ▶️Наличие chunks - блоков, которые обрабатываются параллельно. Это ускоряет вычисления и снижает нагрузку на RAM. Эти блоки можно указывать как и явно (через кол-во или размер), так и по умолчанию через auto. ▶️Необходимость явно вызывать вычисления при помощи .compute(). Всё, написанное до .compute(), только лишь задается, а не фактически вычисляется. Сделано это для оптимизации при работе с большими данными

В дополнение к прошлому посту - вывод результатов EXPLAIN PLAN

Основы анализа запросов в SQL на производительность Такой анализ делается с помощью EXPLAIN PLAN (говорилось ранее). Применяется следующим образом: EXPLAIN PLAN FOR select u.*, u2.email from users u left join users_2 u2 on u.user_id = u2.user_id; SELECT * FROM TABLE(DBMS_XPLAN.DISPLAY); Вывод для такого запроса в следующем посте ⬇️ 🟠На что обращать внимание при выводе результатов EXPLAIN PLAN? Есть такие параметры как: ▶️ID - номер шага. Определяет порядок выполнения операций. Вложенные операции имеют больший ID и сдвинуты правее. Также по ID можно отслеживать родительско-дочерние отношения между операциями ▶️Operation - показывает, какие операции выполняет СУБД. ▶️Name - имя объекта (таблицы, индекса, представления), к которому обращается операция ▶️Rows — оценка количества строк ▶️Bytes — объем данных ▶️Cost — затраты на CPU ▶️Time — оценка времени выполнения 🟠В данном примере можно рассуждать так: SELECT STATEMENT - это корневая операция плана выполнения, представляющая весь SQL-запрос. Все другие операции являются дочерними по отношению к ней. Поэтому она идет под ID = 0. Основная операция в этом запросе - HASH JOIN OUTER. Внешнее соединение (OUTER) означает, что будут возвращены все строки из левой таблицы (USERS), даже если нет соответствия в правой (USERS_2). В данном примере использование именно HASH JOIN - своего рода проблема, которую можно было бы подкорректировать с помощью NESTED LOOPS, т.к. сами таблицы достаточно маленькие. Но в данном случае, когда таблицы ну уж очень маленькие, запрос выполнится быстро и без этого. Остальные операции - это 2 TABLE ACCESS FULL - то есть полное сканирование таблицы. Опять же, в данном случае из-за размера таблиц в этом нет ничего такого. Но вообще это - одна из самых частых причин медленной работы SQL-запросов. Если бы таблицы были большие, то желательно бы добавить индексы на столбцы соединения. Теперь касательно Rows. Весь запрос вернёт 4 строки (ID=0). А именно: джоин даст 4 строки (ID=1), т.к. используется left join по отношению к USERS, которая, свою очередь, даст эти же 4 строки (ID=2). Таблица USERS_2 же вернет 2 строки (ID=3) после джоина. Bytes - просто ожидаемый объём данных в байтах, который вернёт операция. Cost (%CPU) - использование CPU. В данном примере общая стоимость запроса = 4% от CPU. 🟠Ну а вообще, ключевые моменты, на которые надо обращать внимание при анализе производительности типовых, не самых заумных SQL-запросов, такие: ▶️Устранять TABLE ACCESS FULL с помощью индексов, если эта операция выполняется ощутимо долго. Можно еще уменьшить размер таблицы с помощью партиционирования. ▶️Если используются джоины, то понимать, где лучше применять HASH JOIN, а где NESTED LOOPS. ▶️Также потенциально медленными могут быть операции сортировки (где будет написано SORT). Решается опять же с помощью индексов. ▶️Операции FILTER могут быть проблемными, особенно если процент отфильтрованных строк слишком высок. ▶️Еще VIEW-операции могут выполняться долго. Основной совет - не путать VIEW с обычной таблицей и не особо нагружать дополнительными функциями те запросы, где фигурируют вьюхи. Особенно если еще создается view на view. ▶️Ну и смотреть на кол-во ожидаемых возвращаемых строк, байтов, смотреть на Cost (%CPU) и, само собой, на время выполнения для понимания того, какая из операций наиболее тяжеловесна и что именно надо оптимизировать.

Hints (Хинты) в SQL Хинты - это специальные "подсказки", которые добавляются в SQL-запросы, чтобы влиять на поведение оптимизатора СУБД. Они позволяют помочь СУБД в выборе наиболее производительного варианта выполнения запроса в случае, если способ по умолчанию кажется не самым оптимальным. Хинты прописываются сразу после SELECT следующим образом: SELECT /*+ HINT */ column_1, column_2... Например, как я уже упоминал ранее, таким способом можно влиять на выполнение индексов в SQL-запросе. Условно, если мы создали такой индекс CREATE INDEX idx_user_name ON users (name) который при выполнении такого запроса SELECT * FROM USERS WHERE NAME = 'IVAN' не хочет выполняться, то можно СУБД заставить это сделать: SELECT /*+ INDEX(users idx_user_name) */ * FROM USERS WHERE NAME = 'IVAN' Удостовериться в выполнении индекса, да и вообще в целом посмотреть, работают ли хинты и по какой логике выполняется тот или иной запрос, можно с помощью EXPLAIN PLAN: EXPLAIN PLAN FOR SELECT /*+ INDEX(users idx_user_name) */ * FROM USERS WHERE NAME = 'IVAN'; --Отображение результата (oracle): SELECT * FROM TABLE(DBMS_XPLAN.DISPLAY) -- или DBMS_XPLAN.DISPLAY_CURSOR Выполнение такого запроса выведет таблицу с информацией о том, выполняются ли хинты, какие операции задействованы в запросе, какая у них производительность и т.п. Подробнее об EXPLAIN PLAN и о том, как работать с такой таблицей, я расскажу чуть позже. Бывает много разных хинтов, вот некоторые из них: ▶️/*+ INDEX(table index) */ Принудительное использование индекса ▶️/*+ NO_INDEX(table index) */ Запрет использования индекса ▶️/*+ USE_NL(table_1 table_2) */ Вложенные циклы (Nested Loops). Может ускорить работу джоинов, когда одна таблица очень маленькая ▶️/*+ USE_HASH(table_1 table_2) */ Хэш-соединение. Может ускорить работу джоинов, когда таблицы большие ▶️/*+ PARALLEL(table) */ Параллельное выполнение. Может ускорить выполнение запроса в обмен на более высокое использование CPU. ▶️/*+ NO_PARALLEL(table) */ Запрет на параллельное выполнение. Напротив, экономится CPU, но запрос может выполняться медленнее. ▶️/*+ FIRST_ROWS(n) */ Позволяет быстрее возвращать первые строки, но может увеличить общее время выполнения. Работает, если до этого создан индекс. ▶️/*+ CACHE(table) */ Кэширование таблицы в памяти. Если таблица часто используется, то это может помочь быстрее получать к ней доступ. ▶️/*+ NOCACHE(table) */ Отмена кэширования таблицы. Если таблица большая и редко используется, то можно запретить её кэшировать для экономии памяти.

Pivot в SQL Достаточно любопытная и полезная функция, которая позволяет создавать сводные таблицы в SQL. Синтаксис в общем виде: SELECT * FROM ( SELECT столбец_1, столбец_2... FROM табл. WHERE условия ) PIVOT ( агрегатная функция(столбец) FOR столбец_2 IN (список значений для столбцов) ) ORDER BY столбцы сортировки То есть. Например, есть таблица sales (табл. 1). И надо посчитать сумму продаж и средний чек по продуктам в разрезе регионов. Для этого селектим нужные столбцы (prod_id, amount, region). Для наглядности можно ограничить prod_id с 4 по 7. Далее создается пивот: пишутся агрегатные функции на amount (в нашем случае SUM и AVG), а затем для значений в region создаются отдельные столбцы (в FOR). Можно это все отсортировать. Готовый запрос: SELECT * FROM ( SELECT product_id, region, amount FROM sales WHERE product_id IN (4, 5, 6, 7) ) PIVOT ( SUM(amount) AS sum, AVG(amount) AS avg FOR region IN ('north' AS north, 'south' AS south) ) ORDER BY product_id Итог на табл. 2

Слои данных в DWH В архитектуре хранилищ данных (Data Warehouse) обычно выделяют несколько логических слоев, каждый из которых выполняет свою функцию в процессе обработки и хранения информации. Вообще существует много разных точек зрения на то, что из себя должны представлять слои DWH и сколько их должно быть. Раскрою одну из типовых: 1️⃣Staging. Это начальный уровень в архитектуре хранилища данных, куда поступает первичная информация из разнообразных источников. Проще говоря, это сырые данные, которые могут быть плохо адаптированы под общую логику хранилища и даже содержать разные ошибки. Загружаются они сюда из разных источников как есть, без изменений. 2️⃣ODS (Operational Data Store). Хранилище для операционных данных. На этом этапе данные уже адаптируются под общую логику хранилища, исправляются ошибки, данные приобретают единый формат, заполняются недопустимые пустоты и т.п. Важно то, что это - именно текущие данные, а не исторические. 3️⃣DDS (Detail Data Store). Хранилище для детальных исторических данных. Это, можно сказать, ключевой слой в хранилище, здесь данные отражены в наиболее полной мере. Загружаясь из ODS, данные разбиваются на таблицы. Именно здесь (как правило) применяются различные схемы DWH (подробнее про схемы писал здесь и здесь). 4️⃣Data Marts. Слой витрин данных, где информация структурирована и готова к использованию бизнес-аналитиками, делается так называемый срез данных из хранилища. Подробнее об этом писал здесь. 5️⃣REP (Reporting), или Aggregations. Это полностью готовые данные, представленные в удобном для визуализации и анализа виде, которые используются уже для понимания итогов. 🟠Можно также выделить и дополнительные слои: ▶️Sandbox (Песочница). Для экспериментального анализа данных. Грубо говоря, изолированный слой-черновик, где можно делать что угодно без риска сломать основное хранилище. ▶️Data Lake. Слой с необработанными первичными данными, такой же как и Staging. Но от Staging всё же отличается: - Staging хранит данные временно, а Data Lake - постоянно - Staging хранит данные, которые поступят в хранилище дальше, а Data Lake хранит в том числе и постороннюю информацию - Staging хранит только структурированные (иногда полуструктурированные) данные, а Data Lake - что угодно (JSON, CSV, изображения, видео и т.п.) - Основное назначение Staging - ETL-процессы в слой ODS; основное назначение Data Lake - разные данные для аналитики ▶️Semantic Layer (Семантический слой) Слой, предоставляющий бизнес-термины вместо технических названий таблиц и полей. Позволяет работать с данными, используя привычные бизнес-понятия вместо сложных SQL-запросов. Например, не sales, а "Продажи", не SUM(amount) / COUNT(order_id), а "Средний чек" и т.п.

Представления (Views) в SQL Представления (views) - это виртуальные таблицы в базе данных, которые не хранят данные физически, а представляют собой сохраненные SQL-запросы. Нужны они для упрощения организации работы с запросами. Условно, есть у нас несколько таблиц: продажи, клиенты, магазины, продавцы. И нам надо ежемесячно собирать эти все данные воедино с помощью джоинов и анализировать всё это. На помощь приходят вьюшки: так как каждый месяц это всё происходит снова и снова, то можно написать основную логику по агрегации этих таблиц с помощью запросов, сохранить этот запрос в представление и обращаться к нему 1 строкой кода. При необходимости можно что-то добавлять, фильтровать, сортировать и т.п. Также еще немаловажно то, что, если рассматривать вью как таблицу, а не как сохраненный SQL-запрос, то эту таблицу не надо обновлять - это всё происходит само собой, когда изменяются данные в тех таблицах, на которые она опирается. 🟠Пример создания view: CREATE VIEW active_customers AS SELECT id, name, email FROM customers WHERE status = 'active'; После выполнения view сохраняется в базе данных. Вызвать ее можно, например, вот так: select * from active_customers Такой запрос вернет поля id, name, email из таблицы customers, где status = 'active'. Проще говоря, по своей сути это напоминает СТЕ таблицы, только сохраненные в базе данных. Найти созданные представления можно через интерфейс СУБД, либо из таких таблиц, как user_views, all_views и т.д.: SELECT * FROM user_views where view_name = 'ACTIVE_CUSTOMERS' 🟠Также представления допускают операции INSERT, UPDATE, DELETE. Как пример: UPDATE active_customers SET email = '[email protected]' WHERE id = 2 После такого запроса обновится в первую очередь таблица, на которой построено представление, и, как следствие, само представление. 🟠Специфика представлений: ▶️Нужно помнить, что в первую очередь view - это сохраненный SQL-запрос, который выполняется при обращении к ней. Поэтому вызов представления - процесс намного более медленный, чем просто выбор данных из обычной таблицы. ▶️С некоторыми представлениями будет проблематично взаимодействовать с помощью INSERT, UPDATE, DELETE. Как правило это те, которые используют JOIN, GROUP BY, DISTINCT, UNION... Связано это банально с тем, что просто непонятно, что именно и где именно обновлять/удалять. ▶️Представления зависят от базовых таблиц, и при изменении структуры таблиц view может стать неработоспособным

Индексы в SQL Индексы в SQL — это структуры данных, которые ускоряют поиск и доступ к строкам в таблице. С первого взгляда они могут показаться похожими на первичные и внешние ключи, а также на встроенные функции нумерации строк в таблице (например row_number). На деле же это совершенно разные вещи. В чем же их различие? ▶️Главная задача первичных ключей - обеспечение уникальности и идентификация каждой строки в таблице. ▶️Главная задача внешних ключей - связать таблицы между собой. ▶️Главная задача разных функций по типу row_number() - присвоение какого-либо номера каждой строке для разных целей, например, для дальнейшего удобства работы со строкой в запросах в СТЕ таблицах. ▶️И, наконец, одна из основных задач индексов - обеспечение более быстрого доступа к данным. Иными словами - оптимизация. 🟠Как это работает? Например, есть у нас таблица USERS с полями user_id, name, age, email. В этой таблице мы часто выполняем поиск по полю name: SELECT * FROM USERS WHERE NAME = 'IVAN' Чтобы сделать этот поиск в следующие разы быстрее, можно создать простой индекс: CREATE INDEX idx_user_name ON users (name) После того, как такой запрос выполнится, индекс просто сохранится в базу данных (прямо как таблица после CREATE TABLE, только немного в другое место и в другом виде) Найти его можно таким запросом: SELECT index_name, table_name, uniqueness FROM user_indexes --или например ALL_INDEXES WHERE index_name = 'IDX_USER_NAME'; После того, как индекс сохранился в БД, делать с ним ничего не надо. СУБД автоматически решает, использовать ли индекс, основываясь на оптимизаторе запросов и статистике. Можно, конечно, и самому повлиять на использование индекса, но это уже отдельная история. ✅Индекс, вероятнее всего, будет использован для ускорения запроса, если в запросе используется условие WHERE с точным значением для индексированного столбца, а также >,