Создание конвейеров данных ETL с помощью Pandas

# Один и тот же интерфейс для разных источников
sales_data = pd.read_csv('sales.csv')
customer_data = pd.read_excel('customers.xlsx')
api_response = pd.read_json('api_data.json')
 
# Объединение данных из разных источников
combined_data = sales_data.merge(customer_data, on='customer_id')

# Обнаружение пропусков
missing_values = data.isnull().sum()
 
# Умная заполнение пропусков 
data.fillna({'numeric_col': data['numeric_col'].median(), 
             'category_col': 'Unknown'}, inplace=True)
             
# Удаление дубликатов
data.drop_duplicates(subset=['transaction_id'], inplace=True)

# Преобразование из "широкого" в "длинный" формат
melted_data = pd.melt(wide_data, 
                       id_vars=['region', 'product'], 
                       value_vars=['Q1', 'Q2', 'Q3', 'Q4'],
                       var_name='quarter', 
                       value_name='sales')

# Группировка, агрегация и сортировка одним махом
report = (data.groupby(['region', 'product_category'])
              .agg({'sales': ['sum', 'mean', 'count'], 
                    'profit': ['sum', 'mean']})
              .sort_values(('sales', 'sum'), ascending=False))

# Анализ транзакций: найти топ-5 клиентов по общей сумме покупок в 2022 году,
# но только по транзакциям больше 1000 руб.
result = (transactions
         .query('date >= "2022-01-01" and date <= "2022-12-31" and amount > 1000')
         .groupby('customer_id')
         .agg({'amount': 'sum'})
         .sort_values('amount', ascending=False)
         .head(5))

# От анализа к визуализации и ML за три строки
processed_data = df.dropna().transform(normalize_features)
plt.figure(figsize=(10, 6))
sns.heatmap(processed_data.corr(), annot=True, cmap='coolwarm')
model = RandomForestClassifier().fit(processed_data.drop('target', axis=1), processed_data['target'])

# Классический ETL-подход в современном исполнении на Pandas
# Этап 1: Извлечение
raw_data = pd.read_csv('source_system_dump.csv')
 
# Этап 2: Трансформация
transformed_data = (raw_data
                   .drop_duplicates()
                   .fillna(0)
                   .query('revenue > 0')
                   .assign(profit_margin = lambda x: x['profit'] / x['revenue']))
 
# Этап 3: Загрузка
transformed_data.to_sql('analytics_table', db_connection)

# Современный гибридный подход: объединение потоковых и пакетных данных
batch_data = pd.read_parquet('hourly_aggregates.parquet')
stream_data = pd.DataFrame(kafka_consumer.poll(10000))
 
# Синхронизация и объединение
combined_analytics = pd.concat([batch_data, stream_data]).drop_duplicates()

# Типичный pre-processing для ML с использованием Pandas
def prepare_features(df):
    # Обработка пропусков
    df = df.fillna(df.median(numeric_only=True))
    
    # One-hot encoding для категориальных переменных
    df = pd.get_dummies(df, columns=['category', 'region'])
    
    # Нормализация числовых признаков
    for col in df.select_dtypes('number').columns:
        df[col] = (df[col] - df[col].min()) / (df[col].max() - df[col].min())
    
    return df
 
model_ready_data = prepare_features(raw_data)

# Обогащение данных предсказаниями модели
predictions = model.predict(features)
enriched_data = raw_data.copy()
enriched_data['predicted_class'] = predictions
enriched_data['confidence_score'] = model.predict_proba(features).max(axis=1)

# Классический ETL-пайплайн на Pandas
# 1. Extract: Извлечение данных из разных источников
users = pd.read_csv('users.csv')
transactions = pd.read_json('transactions.json')
web_logs = pd.read_parquet('web_activity.parquet')
 
# 2. Transform: Очистка и преобразование
users = users.drop_duplicates(subset=['user_id'])
transactions = transactions[transactions['status'] == 'completed']
user_transactions = pd.merge(users, transactions, on='user_id', how='left')
 
# Обогащение логами активности
enriched_data = pd.merge(user_transactions, web_logs, on='user_id', how='left')
enriched_data['revenue_per_visit'] = enriched_data['amount'] / enriched_data['visit_count']
 
# 3. Load: Загрузка в хранилище для анализа
enriched_data.to_sql('analytics_mart', sql_connection)

# Современный ELT-пайплайн
# 1. Extract & Load: Сырые данные сразу отправляются в хранилище
raw_users = pd.read_csv('users.csv')
raw_users.to_parquet('datalake/users/date=20230621/raw.parquet')
 
# 2. Transform: происходит при необходимости анализа
def transform_for_analysis():
    raw_files = glob.glob('datalake/users/*/raw.parquet')
    combined = pd.concat([pd.read_parquet(f) for f in raw_files])
    
    # Очистка и трансформации по требованию
    return combined.drop_duplicates().fillna(0)

def validate_schema(df):
    expected_columns = ['user_id', 'timestamp', 'event_type']
    missing = set(expected_columns) - set(df.columns)
    if missing:
        raise ValueError(f"Отсутствуют столбцы: {missing}")
    return df
 
def handle_duplicates(df):
    initial_rows = len(df)
    df = df.drop_duplicates()
    print(f"Удалено {initial_rows - len(df)} дубликатов")
    return df
 
# Построение конвейера через композицию функций
processed_data = (raw_data
                 .pipe(validate_schema)
                 .pipe(handle_duplicates)
                 .pipe(lambda df: df[df['event_type'].isin(['purchase', 'view'])]))

def process_transaction_batch(batch):
    # Обработка очередной порции транзакций
    df = pd.DataFrame(batch)
    
    # Быстрая фильтрация и обогащение
    df = df.query('amount > 0').copy()
    df['timestamp'] = pd.to_datetime(df['timestamp'])
    df['hour'] = df['timestamp'].dt.hour
    
    # Инкрементальное обновление аналитических метрик 
    update_metrics(df)
    return df
 
# Обработка потоковых данных порциями
for batch in stream_source.fetch_batches(batch_size=10000):
    processed = process_transaction_batch(batch)
    if len(processed) > 0:
        send_to_monitoring_service(processed)

class LazyTransformer:
    def __init__(self, source_path):
        self.source = source_path
        self.transformations = []
        self._data = None
    
    def add_transformation(self, func):
        self.transformations.append(func)
        return self
    
    def execute(self):
        if self._data is None:
            self._data = pd.read_csv(self.source)
            
        for transform in self.transformations:
            self._data = transform(self._data)
        
        return self._data
 
# Использование
pipeline = (LazyTransformer('huge_dataset.csv')
           .add_transformation(lambda df: df.dropna())
           .add_transformation(lambda df: df[df['value'] > 0]))
 
# Данные не загружаются, пока явно не запросим результат
result = pipeline.execute()

# Гибридный подход: SQL + Pandas
query = """
    SELECT 
        date, 
        product_category,
        SUM(sales) as total_sales,
        COUNT(DISTINCT customer_id) as unique_customers
    FROM sales_table
    WHERE date >= '2023-01-01'
    GROUP BY date, product_category
"""
# SQL делает тяжелую работу с данными в БД
base_aggregation = pd.read_sql(query, database_connection)
 
# Pandas берет на себя финальные трансформации и подготовку к визуализации
final_report = (base_aggregation
               .pivot(index='date', columns='product_category', values='total_sales')
               .fillna(0)
               .rolling(window=7).mean())

# Компонент извлечения из разных источников
sources = {
    'customers': pd.read_excel('customers.xlsx'),
    'transactions': pd.read_json('api_response.json'),
    'product_catalog': pd.read_sql('SELECT * FROM products', db_connection),
    'real_time_events': pd.DataFrame(kafka_consumer.poll(timeout=5000))
}

# Компонент валидации
def validate_with_metrics(df, validation_rules):
    metrics = {'total_rows': len(df), 'failed_validations': {}}
    
    for rule_name, condition in validation_rules.items():
        invalid_mask = ~df.eval(condition)
        invalid_count = invalid_mask.sum()
        
        if invalid_count > 0:
            metrics['failed_validations'][rule_name] = invalid_count
            # Логирование проблемных записей для дальнейшего анализа
            problem_rows = df[invalid_mask]
            log_validation_issues(rule_name, problem_rows)
    
    return metrics

# Взаимосвязь компонентов через метаданные
class PipelineStage:
    def __init__(self, transform_func):
        self.transform = transform_func
        self.metadata = {}
    
    def process(self, df, upstream_metadata=None):
        # Получение метаданных от предыдущего этапа
        if upstream_metadata:
            self.metadata['upstream'] = upstream_metadata
        
        # Выполнение трансформации
        start_time = time.time()
        result = self.transform(df)
        
        # Фиксация собственных метаданных
        self.metadata['execution_time'] = time.time() - start_time
        self.metadata['output_rows'] = len(result)
        self.metadata['output_columns'] = list(result.columns)
        
        return result, self.metadata

# data_pipeline.py
import pandas as pd
import datetime as dt
 
def run_daily_pipeline():
    # Извлечение данных за последний день
    yesterday = dt.datetime.now() - dt.timedelta(days=1)
    date_str = yesterday.strftime('%Y-%m-%d')
    
    # Загрузка и обработка
    new_data = pd.read_csv(f'daily_logs_{date_str}.csv')
    processed = preprocess_pipeline(new_data)
    
    # Объединение с историческими данными
    historical = pd.read_parquet('historical_data.parquet')
    updated = pd.concat([historical, processed]).drop_duplicates()
    
    # Сохранение обновленного набора
    updated.to_parquet('historical_data.parquet')
    
    # Генерация отчетов
    generate_reports(updated)
    
if __name__ == '__main__':
    run_daily_pipeline()

# Запуск в 3 часа ночи каждый день
0 3 * * * /usr/bin/python3 /path/to/data_pipeline.py

# airflow_dag.py
from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from datetime import datetime, timedelta
 
def extract_data([B]context):
    date = context['execution_date']
    data = pd.read_csv(f's3://bucket/logs/{date.strftime("%Y/%m/%d")}/events.csv')
    # Сохраняем промежуточный результат
    data.to_parquet('/tmp/extracted_data.parquet')
 
def transform_data([/B]context):
    data = pd.read_parquet('/tmp/extracted_data.parquet')
    # Трансформации на Pandas
    processed = data.groupby('user_id').agg({'event': 'count', 'revenue': 'sum'})
    processed.to_parquet('/tmp/processed_data.parquet')
 
def load_data(**context):
    data = pd.read_parquet('/tmp/processed_data.parquet')
    data.to_sql('analytics_table', db_connection, if_exists='append')
 
default_args = {
    'owner': 'data_team',
    'depends_on_past': False,
    'start_date': datetime(2023, 1, 1),
    'retries': 3,
    'retry_delay': timedelta(minutes=5),
}
 
with DAG('daily_analytics_pipeline', default_args=default_args, schedule_interval='@daily') as dag:
    extract_task = PythonOperator(
        task_id='extract_data',
        python_callable=extract_data,
        provide_context=True,
    )
    
    transform_task = PythonOperator(
        task_id='transform_data',
        python_callable=transform_data,
        provide_context=True,
    )
    
    load_task = PythonOperator(
        task_id='load_data',
        python_callable=load_data,
        provide_context=True,
    )
    
    # Определяем последовательность выполнения
    extract_task >> transform_task >> load_task

# Мониторинг критических метрик данных
def monitor_dataframe_state(df, stage_name):
    metrics = {
        'timestamp': datetime.now().isoformat(),
        'stage': stage_name,
        'row_count': len(df),
        'column_count': len(df.columns),
        'missing_values': df.isna().sum().sum(),
        'memory_usage': df.memory_usage(deep=True).sum() / (1024 * 1024),  # МБ
        'numeric_columns_stats': {}
    }
    
    # Собираем статистики по числовым столбцам
    for col in df.select_dtypes('number').columns:
        metrics['numeric_columns_stats'][col] = {
            'min': df[col].min(),
            'max': df[col].max(),
            'mean': df[col].mean(),
            'median': df[col].median()
        }
    
    # Сохраняем метрики для дальнейшего анализа
    with open(f'pipeline_metrics_{stage_name}.json', 'a') as f:
        f.write(json.dumps(metrics) + '\n')
    
    # Проверяем на аномалии
    if metrics['row_count'] == 0:
        raise ValueError(f"Этап {stage_name} вернул пустой DataFrame!")
    
    return df

def debug_transform(input_df, transform_func, debug_dir='debug_snapshots'):
    """Обертка для отладки трансформаций с сохранением промежуточных состояний"""
    os.makedirs(debug_dir, exist_ok=True)
    
    # Сохраняем входные данные
    timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
    input_path = f"{debug_dir}/{timestamp}_input.pkl"
    input_df.to_pickle(input_path)
    
    try:
        # Выполняем трансформацию
        result = transform_func(input_df)
        
        # Сохраняем результат
        output_path = f"{debug_dir}/{timestamp}_output.pkl"
        result.to_pickle(output_path)
        
        print(f"Отладочные данные сохранены: {input_path}, {output_path}")
        return result
    except Exception as e:
        print(f"Ошибка в трансформации: {str(e)}")
        print(f"Входные данные сохранены в {input_path}")
        raise

def profile_pipeline_stages(data, stages):
    """Профилирование каждого этапа конвейера"""
    result = data.copy()
    timings = {}
    
    for stage_name, stage_func in stages:
        start_time = time.time()
        result = stage_func(result)
        execution_time = time.time() - start_time
        
        timings[stage_name] = {
            'time': execution_time,
            'rows': len(result),
            'speed': len(data) / execution_time if execution_time > 0 else 0
        }
        
        print(f"Этап {stage_name}: {execution_time:.2f} сек, {len(result)} строк")
    
    return result, timings

# Регрессионное тестирование с оповещениями
def check_for_regressions(current_metrics, historical_metrics, threshold=0.2):
    """Проверка на серьезные отклонения от исторических значений"""
    alerts = []
    
    for metric, current_value in current_metrics.items():
        if metric not in historical_metrics:
            continue
            
        historical = historical_metrics[metric]
        percent_change = abs(current_value - historical) / historical
        
        if percent_change > threshold:
            alerts.append(f"Регрессия в метрике {metric}: изменение на {percent_change:.1%}")
    
    if alerts:
        send_alert_email("\n".join(alerts))

class DataSourceFactory:
    @staticmethod
    def get_source(source_type, **params):
        if source_type == 'csv':
            return lambda: pd.read_csv(params['path'])
        elif source_type == 'database':
            return lambda: pd.read_sql(params['query'], params['connection'])
        elif source_type == 'api':
            return lambda: pd.DataFrame(requests.get(params['url']).json())
            
# Использование            
sources_config = [
    {'type': 'csv', 'name': 'sales', 'path': 'sales.csv'},
    {'type': 'database', 'name': 'customers', 'query': 'SELECT * FROM customers', 'connection': db_conn}
]
 
data_sources = {cfg['name']: DataSourceFactory.get_source(cfg['type'], **cfg) for cfg in sources_config}
sales_data = data_sources['sales']()

class DataProcessor:
    def __init__(self):
        self.next_processor = None
        
    def set_next(self, processor):
        self.next_processor = processor
        return processor
        
    def process(self, data):
        result = self.do_processing(data)
        if self.next_processor:
            return self.next_processor.process(result)
        return result
        
    def do_processing(self, data):
        # Переопределяется в наследниках
        return data
 
# Конкретные обработчики
class MissingValueProcessor(DataProcessor):
    def do_processing(self, data):
        return data.fillna(0)
        
class OutlierProcessor(DataProcessor):
    def do_processing(self, data):
        # Удаление выбросов по z-score
        return data[np.abs((data - data.mean()) / data.std()) < 3]

class PipelineSubject:
    def __init__(self):
        self.observers = []
        
    def attach(self, observer):
        self.observers.append(observer)
    
    def notify(self, event_type, data=None):
        for observer in self.observers:
            observer.update(event_type, data)
            
class MetricsCollector:
    def update(self, event_type, data):
        if event_type == 'stage_completed':
            print(f"Этап {data['stage']} завершен, обработано {len(data['dataframe'])} строк")
            # Сохранение метрик в базу...

def process_chunk(chunk):
    # Обработка отдельного фрагмента данных
    return chunk.groupby('category').sum()
 
# Разбиение датафрейма на части и параллельная обработка
def parallel_process(df, func, n_workers=4):
    chunks = np.array_split(df, n_workers)
    with concurrent.futures.ProcessPoolExecutor(max_workers=n_workers) as executor:
        results = list(executor.map(func, chunks))
    return pd.concat(results)

# До оптимизации
print(df.info())
[H2]Int64, Float64 для чисел и объекты для строк[/H2]
 
# После оптимизации
optimized_df = df.copy()
# Понижаем точность вещественных чисел
for col in df.select_dtypes('float64').columns:
    optimized_df[col] = df[col].astype('float32')
    
# Используем категориальные типы для строк с повторами
for col in df.select_dtypes('object').columns:
    if df[col].nunique() / len(df) < 0.5:  # Много повторов
        optimized_df[col] = df[col].astype('category')
 
print(optimized_df.info())
# Сокращение памяти часто на 30-70%

# Обработка крупного файла кусками
chunk_size = 100000
chunks = []
 
for chunk in pd.read_csv('massive_log.csv', chunksize=chunk_size):
    # Обрабатываем каждый кусок отдельно
    processed_chunk = process_data(chunk)
    
    # Сохраняем только нужные результаты агрегации
    summary = processed_chunk.groupby('category').agg({'value': ['sum', 'mean']})
    chunks.append(summary)
 
# Объединяем результаты
final_result = pd.concat(chunks)
final_summary = final_result.groupby(level=0).sum()  # Если нужно суммировать результаты групп

# Медленный способ с циклом
def slow_process(df):
    result = []
    for idx, row in df.iterrows():
        result.append(complex_calculation(row['a'], row['b']))
    df['result'] = result
    return df
 
# Быстрый векторизованный способ
def fast_process(df):
    df['result'] = complex_calculation_vectorized(df['a'], df['b'])
    return df

# Инвертируем логику конвейера
def optimized_pipeline(data_source):
    # Читаем только нужные столбцы
    df = pd.read_csv(data_source, usecols=['id', 'timestamp', 'value'])
    
    # Фильтруем как можно раньше
    df = df[df['value'] > 0]
    
    # Только потом делаем тяжелые операции с уменьшенным набором
    return heavy_processing(df)

import pandas as pd
import numpy as np
from multiprocessing import Pool, cpu_count
 
def process_group(group_data):
    # Тяжелые вычисления для одной группы
    return group_data.assign(processed_value=complex_calculation(group_data))
 
def parallel_process(df, group_column):
    # Разбиение данных по группам
    groups = [group for _, group in df.groupby(group_column)]
    
    # Параллельная обработка
    with Pool(processes=cpu_count()) as pool:
        results = pool.map(process_group, groups)
    
    # Объединение результатов
    return pd.concat(results)

import dask.dataframe as dd
 
# Преобразование Pandas DataFrame в Dask DataFrame
dask_df = dd.from_pandas(big_pandas_df, npartitions=cpu_count()*2)
 
# Выполнение тех же операций, что и в Pandas, но параллельно
result = dask_df.groupby('category').agg({'value': 'mean'}).compute()

from pandarallel import pandarallel
pandarallel.initialize()
 
# Обычный apply работает на одном ядре
[H2]df['result'] = df.apply(heavy_function, axis=1)[/H2]
 
# Распараллеленный вариант использует все ядра
df['result'] = df.parallel_apply(heavy_function, axis=1)

# До оптимизации
df = pd.read_csv('huge_dataset.csv')
print(df.info(memory_usage='deep'))  # Показывает потребление памяти
 
# Понижение точности типов данных
def reduce_memory_usage(df):
    for col in df.columns:
        if df[col].dtype == 'int64':
            df[col] = pd.to_numeric(df[col], downcast='integer')
        elif df[col].dtype == 'float64':
            df[col] = pd.to_numeric(df[col], downcast='float')
    return df
 
optimized_df = reduce_memory_usage(df)
print(optimized_df.info(memory_usage='deep'))  # Сравниваем результат

# Преобразование текстовых столбцов в категории
for col in df.select_dtypes('object').columns:
    if df[col].nunique() / len(df) < 0.5:  # Если уникальных значений менше 50%
        df[col] = df[col].astype('category')

# Не загружаем лишние столбцы
needed_columns = ['user_id', 'timestamp', 'event_type']
df = pd.read_csv('huge_logs.csv', usecols=needed_columns)

# Освобождаем память после использования
del temporary_df
import gc
gc.collect()  # Принудительный сбор мусора

# Чтение данных напрямую из AWS S3
import boto3
import pandas as pd
import io
 
s3_client = boto3.client('s3')
response = s3_client.get_object(Bucket='my-analytics-bucket', Key='reports/daily_logs.csv')
df = pd.read_csv(io.BytesIO(response['Body'].read()))
 
# Запись обработанных данных обратно в S3
buffer = io.StringIO()
processed_df.to_csv(buffer, index=False)
s3_client.put_object(
    Body=buffer.getvalue(),
    Bucket='my-analytics-bucket',
    Key='reports/processed_data.csv'
)

from google.cloud import storage
from io import BytesIO
 
client = storage.Client()
bucket = client.get_bucket('analytics-bucket')
blob = bucket.blob('monthly_report.parquet')
data = BytesIO(blob.download_as_bytes())
df = pd.read_parquet(data)

# Выполнение запроса через AWS Athena и получение результатов в Pandas
import pyathena
 
conn = pyathena.connect(s3_staging_dir='s3://athena-query-results/',
                       region_name='us-west-2')
df = pd.read_sql("SELECT * FROM logs.user_events WHERE date = CURRENT_DATE", conn)

# Запуск PySpark в AWS EMR и перенос результатов в Pandas
from pyspark.sql import SparkSession
 
spark = SparkSession.builder.appName("BigDataProcessing").getOrCreate()
spark_df = spark.read.parquet("s3://big-data-bucket/massive-logs/")
filtered = spark_df.filter("event_type = 'purchase'").groupBy("user_id").count()
 
# Конвертация результатов Spark в Pandas для тонкой настройки
pandas_df = filtered.toPandas()

# Умное чтение только нужных частей данных
import s3fs
fs = s3fs.S3FileSystem()
 
with fs.open('s3://analytics-bucket/huge_partitioned_data/year=2023/month=06/day=15/data.parquet') as f:
    df = pd.read_parquet(f, columns=['user_id', 'purchase_amount'])