Эффективный рабочий процесс анализа данных с использованием Lilac

Введение в создание рабочего процесса анализа данных с помощью Lilac

В современном мире, где данные становятся основой принятия решений, создание эффективного рабочего процесса анализа данных становится критически важным. Библиотека Lilac предлагает мощные инструменты для трансформации, фильтрации и экспорта структурированных данных. В этой статье мы рассмотрим, как построить функциональный рабочий процесс анализа данных, используя Lilac, и как это может помочь вам в вашей работе.

Установка необходимых библиотек

Первым шагом является установка необходимых библиотек. Для этого выполните следующую команду:

!pip install lilac[all] pandas numpy

Эта команда установит полный пакет Lilac, а также библиотеки Pandas и NumPy, которые необходимы для работы с данными.

Импорт библиотек

После установки библиотек, импортируем их в наш проект:

import json
import uuid
import pandas as pd
from pathlib import Path
from typing import List, Dict, Any, Tuple, Optional
from functools import reduce, partial
import lilac as ll

Эти библиотеки помогут нам обрабатывать данные и управлять директориями, а также использовать функциональное программирование для создания чистого и расширяемого кода.

Создание функциональных утилит

Далее определим несколько полезных функций для обработки данных:

def pipe(*functions):
   return lambda x: reduce(lambda acc, f: f(acc), functions, x)

def map_over(func, iterable):
   return list(map(func, iterable))

def filter_by(predicate, iterable):
   return list(filter(predicate, iterable))

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

Генерация тестовых данных

Создадим функцию для генерации тестовых данных, которая будет использоваться в нашем примере:

def create_sample_data() -> List[Dict[str, Any]]:
   return [
       {"id": 1, "text": "Что такое машинное обучение?", "category": "технологии", "score": 0.9, "tokens": 5},
       {"id": 2, "text": "Машинное обучение — это подмножество ИИ", "category": "технологии", "score": 0.8, "tokens": 6},
       {"id": 3, "text": "Свяжитесь с поддержкой для получения помощи", "category": "поддержка", "score": 0.7, "tokens": 4},
       {"id": 4, "text": "Что такое машинное обучение?", "category": "технологии", "score": 0.9, "tokens": 5}, 
       {"id": 5, "text": "Глубокое обучение нейронных сетей", "category": "технологии", "score": 0.85, "tokens": 4},
       {"id": 6, "text": "Как оптимизировать модели?", "category": "технологии", "score": 0.75, "tokens": 5},
       {"id": 7, "text": "Руководство по настройке производительности", "category": "руководство", "score": 0.6, "tokens": 3},
       {"id": 8, "text": "Современные методы оптимизации", "category": "технологии", "score": 0.95, "tokens": 3},
       {"id": 9, "text": "Алгоритм градиентного спуска", "category": "технологии", "score": 0.88, "tokens": 3},
       {"id": 10, "text": "Метрики оценки моделей", "category": "технологии", "score": 0.82, "tokens": 3},
   ]

Эта функция создаёт набор данных, который мы будем использовать для демонстрации возможностей Lilac.

Настройка проекта Lilac

Теперь создадим директорию для нашего проекта:

def setup_lilac_project(project_name: str) -> str:
   project_dir = f"./{project_name}-{uuid.uuid4().hex[:6]}"
   Path(project_dir).mkdir(exist_ok=True)
   ll.set_project_dir(project_dir)
   return project_dir

Эта функция создаёт уникальную директорию для проекта, что позволяет организовать управление файлами данных.

Создание и трансформация наборов данных

Сгенерируем набор данных из тестовых данных:

def create_dataset_from_data(name: str, data: List[Dict]) -> ll.Dataset:
   data_file = f"{name}.jsonl"
   with open(data_file, 'w') as f:
       for item in data:
           f.write(json.dumps(item) + '\n')

   config = ll.DatasetConfig(
       namespace="tutorial",
       name=name,
       source=ll.sources.JSONSource(filepaths=[data_file])
   )

   return ll.create_dataset(config)

Эта функция преобразует тестовые данные в файл JSON Lines и создаёт набор данных для структурированного анализа.

Извлечение и фильтрация данных

Теперь извлечём данные в DataFrame Pandas:

def extract_dataframe(dataset: ll.Dataset, fields: List[str]) -> pd.DataFrame:
   return dataset.to_pandas(fields)

Затем применим функциональные фильтры:

def apply_functional_filters(df: pd.DataFrame) -> Dict[str, pd.DataFrame]:
   filters = {
       'high_score': lambda df: df[df['score'] >= 0.8],
       'tech_category': lambda df: df[df['category'] == 'технологии'],
       'min_tokens': lambda df: df[df['tokens'] >= 4],
       'no_duplicates': lambda df: df.drop_duplicates(subset=['text'], keep='first'),
       'combined_quality': lambda df: df[(df['score'] >= 0.8) & (df['tokens'] >= 3) & (df['category'] == 'технологии')]
   }

   return {name: filter_func(df.copy()) for name, filter_func in filters.items()}

Эти функции позволяют создавать несколько отфильтрованных представлений данных, что облегчает комплексный анализ.

Анализ качества данных

Оценим качество набора данных с помощью следующей функции:

def analyze_data_quality(df: pd.DataFrame) -> Dict[str, Any]:
   return {
       'total_records': len(df),
       'unique_texts': df['text'].nunique(),
       'duplicate_rate': 1 - (df['text'].nunique() / len(df)),
       'avg_score': df['score'].mean(),
       'category_distribution': df['category'].value_counts().to_dict(),
       'score_distribution': {
           'high': len(df[df['score'] >= 0.8]),
           'medium': len(df[(df['score'] >= 0.6) & (df['score'] < 0.8)]),
           'low': len(df[df['score'] < 0.6])
       },
       'token_stats': {
           'mean': df['tokens'].mean(),
           'min': df['tokens'].min(),
           'max': df['tokens'].max()
       }
   }

Эта функция предоставляет важные метрики, позволяя пользователям оценить целостность набора данных и готовность к анализу.

Трансформации и экспорт данных

Определим трансформации для обогащения набора данных:

def create_data_transformations() -> Dict[str, callable]:
   return {
       'normalize_scores': lambda df: df.assign(norm_score=df['score'] / df['score'].max()),
       'add_length_category': lambda df: df.assign(
           length_cat=pd.cut(df['tokens'], bins=[0, 3, 5, float('inf')], labels=['короткий', 'средний', 'длинный'])
       ),
       'add_quality_tier': lambda df: df.assign(
           quality_tier=pd.cut(df['score'], bins=[0, 0.6, 0.8, 1.0], labels=['низкий', 'средний', 'высокий'])
       ),
       'add_category_rank': lambda df: df.assign(
           category_rank=df.groupby('category')['score'].rank(ascending=False)
       )
   }

Применим эти трансформации к DataFrame:

def apply_transformations(df: pd.DataFrame, transform_names: List[str]) -> pd.DataFrame:
   transformations = create_data_transformations()
   selected_transforms = [transformations[name] for name in transform_names if name in transformations]
  
   return pipe(*selected_transforms)(df.copy()) if selected_transforms else df

Наконец, экспортируем отфильтрованные наборы данных в файлы:

def export_filtered_data(filtered_datasets: Dict[str, pd.DataFrame], output_dir: str) -> None:
   Path(output_dir).mkdir(exist_ok=True)

   for name, df in filtered_datasets.items():
       output_file = Path(output_dir) / f"{name}_filtered.jsonl"
       with open(output_file, 'w') as f:
           for _, row in df.iterrows():
               f.write(json.dumps(row.to_dict()) + '\n')
       print(f"Экспортировано {len(df)} записей в {output_file}")

Эта функция организует экспортируемые подмножества, облегчая дальнейшее использование данных.

Основной рабочий процесс анализа

Основная функция координирует полный рабочий процесс:

def main_analysis_pipeline():
   print("Настройка проекта Lilac...")
   project_dir = setup_lilac_project("advanced_tutorial")

   print("Создание тестового набора данных...")
   sample_data = create_sample_data()
   dataset = create_dataset_from_data("sample_data", sample_data)

   print("Извлечение данных...")
   df = extract_dataframe(dataset, ['id', 'text', 'category', 'score', 'tokens'])

   print("Анализ качества данных...")
   quality_report = analyze_data_quality(df)
   print(f"Исходные данные: {quality_report['total_records']} записей")
   print(f"Дубликаты: {quality_report['duplicate_rate']:.1%}")
   print(f"Средний балл: {quality_report['avg_score']:.2f}")

   print("Применение трансформаций...")
   transformed_df = apply_transformations(df, ['normalize_scores', 'add_length_category', 'add_quality_tier'])

   print("Применение фильтров...")
   filtered_datasets = apply_functional_filters(transformed_df)

   print("\nРезультаты фильтрации:")
   for name, filtered_df in filtered_datasets.items():
       print(f"  {name}: {len(filtered_df)} записей")

   print("Экспорт отфильтрованных наборов данных...")
   export_filtered_data(filtered_datasets, f"{project_dir}/exports")

   print("\nЗаписи высшего качества:")
   best_quality = filtered_datasets['combined_quality'].head(3)
   for _, row in best_quality.iterrows():
       print(f"  • {row['text']} (балл: {row['score']}, категория: {row['category']})")

   return {
       'original_data': df,
       'transformed_data': transformed_df,
       'filtered_data': filtered_datasets,
       'quality_report': quality_report
   }

if __name__ == "__main__":
   results = main_analysis_pipeline()
   print("\nАнализ завершён! Проверьте папку экспорта для отфильтрованных наборов данных.")

Этот рабочий процесс демонстрирует интеграцию библиотеки Lilac с принципами функционального программирования, позволяя разработать модульные и выразительные рабочие процессы анализа данных.

Заключение

В результате вы получите практическое понимание создания воспроизводимого рабочего процесса анализа данных, используя абстракции наборов данных Lilac и функциональные программные паттерны для масштабируемого и чистого анализа. Этот туториал охватывает важные этапы, такие как создание наборов данных, трансформация, фильтрация, анализ качества и экспорт, предоставляя гибкость как для экспериментов, так и для развертывания.

Все права на это исследование принадлежат исследователям данного проекта. Для дальнейшего взаимодействия рассмотрите возможность присоединения к профессиональным сообществам или подписки на соответствующие новостные рассылки.