Привет, Хабр!

Сегодня рассмотрим тему автоматизации процессов в хранилищах данных с помощью мощного тандема — Snowflake и Python. Разберем, как с помощью Python можно легко подключаться к Snowflake, загружать данные, управлять таблицами и автоматизировать повседневные задачи.

Настройка среды

Создаем виртуальное окружение для проекта, чтобы изолировать зависимости:

# Для Windows
python -m venv venv
venv\Scripts\activate

# Для macOS/Linux
python3 -m venv venv
source venv/bin/activate

Устанавливаем официальный коннектор Snowflake для Python с помощью pip:

pip install snowflake-connector-python

Затем настраиваем подключение к базе данных Snowflake:

import snowflake.connector

conn = snowflake.connector.connect(
    user='YOUR_USERNAME',
    password='YOUR_PASSWORD',
    account='YOUR_ACCOUNT_IDENTIFIER',
    warehouse='YOUR_WAREHOUSE',
    database='YOUR_DATABASE',
    schema='YOUR_SCHEMA'
)

# курсор для выполнения операций
cursor = conn.cursor()

Помимо коннектора Snowflake, могут понадобиться следующие библиотеки:

pandas;
sqlalchemy и snowflake-sqlalchemy: для использования ORM и облегчения взаимодействия с БД;
python-dotenv: для управления переменными окружения и безопасного хранения учетных данных;
schedule или APScheduler: для планирования и автоматизации задач.

Теперь создаем файл .env и добавляем в него учетные данные:

SNOWFLAKE_USER=YOUR_USERNAME
SNOWFLAKE_PASSWORD=YOUR_PASSWORD
SNOWFLAKE_ACCOUNT=YOUR_ACCOUNT_IDENTIFIER

Затем в скрипте загружаем эти переменные:

import os
from dotenv import load_dotenv

load_dotenv()

user = os.getenv('SNOWFLAKE_USER')
password = os.getenv('SNOWFLAKE_PASSWORD')
account = os.getenv('SNOWFLAKE_ACCOUNT')

#  переменные для подключения
conn = snowflake.connector.connect(
    user=user,
    password=password,
    account=account
)

Синтаксис Python для работы с Snowflake

После настройки окружения и установки необходимых библиотек, можно приступать к работе с Snowflake через Python. Основным инструментом для этого является Snowflake Connector for Python.

Пример подключения и выполнения простого запроса:

import snowflake.connector

# соединение
conn = snowflake.connector.connect(
    user='YOUR_USERNAME',
    password='YOUR_PASSWORD',
    account='YOUR_ACCOUNT_IDENTIFIER'
)

# курсор
cursor = conn.cursor()

# SQL-запрос
cursor.execute("SELECT CURRENT_VERSION()")

# результат
version = cursor.fetchone()
print(f"Текущая версия Snowflake: {version[0]}")

# закрытие соединения
cursor.close()
conn.close()

В этом примере подключаемся к Snowflake, выполняем запрос для получения текущей версии базы данных и выводим результат.

При работе с базами данных важно правильно управлять транзакциями и обрабатывать возможные ошибки. В Snowflake Connector for Python управление транзакциями осуществляется с помощью методов commit() и rollback().

Пример управления транзакциями и обработки исключений:

try:
    # начало транзакции
    conn.cursor().execute("BEGIN")

    # выполнение операций
    cursor.execute("INSERT INTO employees (id, name) VALUES (1, 'Artem')")
    cursor.execute("INSERT INTO employees (id, name) VALUES (2, 'Ivan')")

    # подтверждение транзакции
    conn.cursor().execute("COMMIT")
except Exception as e:
    # откат транзакции в случае ошибки
    conn.cursor().execute("ROLLBACK")
    print(f"Ошибка: {e}")
finally:
    cursor.close()
    conn.close()

В этом коде, если во время выполнения операций произойдет ошибка, транзакция будет откатана, и данные не будут сохранены.

Загрузка данных может быть выполнена с помощью команд PUT и COPY INTO, либо с использованием pandas.

Пример загрузки данных из CSV-файла с помощью COPY INTO:

# загрузка файла в внутренний stage
cursor.execute("""
    PUT file://path/to/data.csv @%your_table
""")

# копирование данных из файла в таблицу
cursor.execute("""
    COPY INTO your_table
    FROM @%your_table/data.csv
    FILE_FORMAT = (TYPE = 'CSV', FIELD_DELIMITER = ',', SKIP_HEADER = 1)
""")

Можно выгрузить данные из таблицы в локальный файл или работать с ними непосредственно в pandas.

Пример выгрузки данных в DataFrame:

import pandas as pd

# запрос и получение данных в DataFrame
df = pd.read_sql("SELECT * FROM your_table", conn)

Используя pandas, можно выполнять различные преобразования данных перед их загрузкой или после выгрузки.

Пример преобразования данных перед загрузкой:

# Чтение данных из CSV
df = pd.read_csv('path/to/data.csv')

# Преобразование данных
df['total_price'] = df['quantity'] * df['unit_price']

# Загрузка данных в Snowflake
from sqlalchemy import create_engine
engine = create_engine('snowflake://...')

df.to_sql('your_table', engine, if_exists='append', index=False)

5 сценариев применения

Автоматическая ежедневная загрузка данных

Допустим, нужно ежедневно загружать новые данные из локального CSV-файла в таблицу Snowflake:

def update_aggregates():
    conn = snowflake.connector.connect(...)
    cursor = conn.cursor()

    # Обновление агрегатов
    cursor.execute("""
        INSERT INTO hourly_aggregates
        SELECT CURRENT_TIMESTAMP, COUNT(*)
        FROM transactions
        WHERE transaction_time >= DATEADD(hour, -1, CURRENT_TIMESTAMP)
    """)

    conn.commit()
    cursor.close()
    conn.close()
    print("Агрегированные данные обновлены.")

# Планирование задачи на каждый час
schedule.every().hour.do(update_aggregates)

Обновление агрегированных данных

Необходимость в ежечасном обновление агрегированные данные для отчетности:

def clean_old_data():
    conn = snowflake.connector.connect(...)
    cursor = conn.cursor()

    # Удаление старых записей
    cursor.execute("""
        DELETE FROM user_activity
        WHERE activity_date < DATEADD(year, -1, CURRENT_DATE)
    """)

    conn.commit()
    cursor.close()
    conn.close()
    print("Старые данные удалены.")

# Планирование задачи на каждое воскресенье в 03:00
schedule.every().sunday.at("03:00").do(clean_old_data)

Мониторинг и уведомления об ошибках

Сценарий: Отслеживать ошибки в процессе ETL и отправлять уведомления ответственному лицу.

Код:

def backup_critical_tables():
    conn = snowflake.connector.connect(...)
    cursor = conn.cursor()

    critical_tables = ['customers', 'orders']
    for table in critical_tables:
        backup_table = f"{table}_backup_{time.strftime('%Y%m%d')}"
        cursor.execute(f"CREATE TABLE {backup_table} CLONE {table}")

    conn.commit()
    cursor.close()
    conn.close()
    print("Резервные копии созданы.")

# Планирование задачи на первое число каждого месяца в 01:00
schedule.every().month.at("01:00").do(backup_critical_tables)

Очистка устаревших данных

Допустим, нужно еженедельно удалять данные старше одного года для оптимизации хранилища:

def backup_critical_tables():
    conn = snowflake.connector.connect(...)
    cursor = conn.cursor()

    critical_tables = ['customers', 'orders']
    for table in critical_tables:
        backup_table = f"{table}_backup_{time.strftime('%Y%m%d')}"
        cursor.execute(f"CREATE TABLE {backup_table} CLONE {table}")

    conn.commit()
    cursor.close()
    conn.close()
    print("Резервные копии созданы.")

# Планирование задачи на первое число каждого месяца в 01:00
schedule.every().month.at("01:00").do(backup_critical_tables)

Автоматическое резервное копирование данных

Сценарий: Ежемесячно создавать резервные копии критически важных таблиц.

Код:

def backup_critical_tables():
    conn = snowflake.connector.connect(...)
    cursor = conn.cursor()

    critical_tables = ['customers', 'orders']
    for table in critical_tables:
        backup_table = f"{table}_backup_{time.strftime('%Y%m%d')}"
        cursor.execute(f"CREATE TABLE {backup_table} CLONE {table}")

    conn.commit()
    cursor.close()
    conn.close()
    print("Резервные копии созданы.")

# Планирование задачи на первое число каждого месяца в 01:00
schedule.every().month.at("01:00").do(backup_critical_tables)

Заключение

Автоматизация процессов в Snowflake с помощью Python открывает широкие возможности для оптимизации работы с хранилищем данных.

Напоминаю про открытый урок «Эффективный анализ данных: Погружение в мир DWH и аналитической инженерии», который пройдет в Otus 23 сентября. На этом уроке участники узнают:

Основы DWH (Data Warehouse): Понимание архитектуры и ключевых компонентов хранилищ данных, а также их роли в бизнес-аналитике.
Инструменты и технологии: Обзор современных инструментов для работы с DWH, таких как ETL-процессы, BI-платформы и языки запросов (SQL).
Практические кейсы: Разбор реальных примеров использования DWH для принятия обоснованных бизнес-решений и оптимизации процессов.

Записаться можно на странице курса «Data Warehouse Analyst».

Автоматизация процессов в DWH с помощью Python и Snowflake