### Install BlossomData Package

Source: https://github.com/azure99/blossomdata/blob/main/README_EN.md

Installs the BlossomData Python package using pip. It also provides an alternative command to install directly from the source repository.

```bash
pip install blossom-data
# Install from source
# pip install git+https://github.com/Azure99/BlossomData.git
```

--------------------------------

### Install BlossomData using pip

Source: https://context7.com/azure99/blossomdata/llms.txt

Installs the BlossomData Python package using pip. This is the primary method for setting up the framework on your local machine.

```bash
pip install blossom-data
```

--------------------------------

### BlossomData Schema Examples

Source: https://github.com/azure99/blossomdata/blob/main/README_EN.md

Illustrates the creation of different Schema types within the BlossomData framework, including TextSchema, ChatSchema, RowSchema, and CustomSchema. These schemas represent various data structures for processing.

```python
# Text data
text_data = TextSchema(content="This is a text content")

# Conversation data
chat_data = ChatSchema(
    messages=[
        user("Hello"),
        assistant("Hello, how can I help you?")
    ]
)

# Structured data
row_data = RowSchema(data={"name": "John", "age": 30, "score": 95})

# Custom data
custom_data = CustomSchema(data=1)
```

--------------------------------

### Distributed Processing with Ray Engine (Python)

Source: https://context7.com/azure99/blossomdata/llms.txt

Utilizes the Ray engine for parallel processing of large datasets. This example demonstrates saving data to a JSONL file, loading it with Ray, repartitioning, executing operators (ChatTranslator, ChatDistiller), shuffling, and writing the processed output. Requires 'blossom' and 'ray' libraries.

```python
from blossom import load_dataset, create_dataset, DatasetEngine
from blossom.op import ChatDistiller, ChatTranslator
from blossom.schema import ChatSchema, user, assistant

# Create sample data
data = [
    ChatSchema(
        id=str(i),
        messages=[user(f"What is {i} + {i}?"), assistant("placeholder")],
    )
    for i in range(1000)
]

# Save to file
create_dataset(data).write_json("large_dataset.jsonl")

# Process with Ray
(
    load_dataset("large_dataset.jsonl", engine=DatasetEngine.RAY)
    .repartition(8)  # Distribute across 8 partitions
    .execute([
        ChatTranslator(model="gpt-4o-mini", target_language="Chinese", parallel=4),
        ChatDistiller(model="gpt-4o-mini", parallel=4),
    ])
    .shuffle()
    .write_json("processed_output")
)

```

--------------------------------

### Distributed Processing with Spark Engine (Python)

Source: https://context7.com/azure99/blossomdata/llms.txt

Leverages Apache Spark for cluster-scale data processing. This example shows initializing a SparkSession, loading data from HDFS, filtering based on metadata, executing a ChatLengthFilter operator, and writing the results back to HDFS. Requires 'pyspark' and 'blossom' libraries.

```python
from pyspark.sql import SparkSession
from blossom import load_dataset, DatasetEngine
from blossom.op import ChatLengthFilter

# Initialize Spark
spark = SparkSession.builder \
    .appName("BlossomData") \
    .master("local[*]") \
    .getOrCreate()

# Process with Spark
(
    load_dataset(
        "hdfs://cluster/data/training_data.jsonl",
        engine=DatasetEngine.SPARK,
        spark_session=spark,
    )
    .filter(lambda x: x.metadata.get("quality", 0) > 0.8)
    .execute([ChatLengthFilter(total_max_len=4096)])
    .write_json("hdfs://cluster/data/filtered_output")
)

spark.stop()

```

--------------------------------

### Create Custom Map Operators with Decorators

Source: https://context7.com/azure99/blossomdata/llms.txt

Demonstrates how to create custom operators for one-to-one transformations using the `@map_operator()` decorator. The example shows converting chat message content to uppercase.

```python
from blossom import create_dataset
from blossom.op import map_operator
from blossom.schema import ChatSchema, ChatRole, user, assistant

data = [
    ChatSchema(messages=[user("Hello"), assistant("Hi!")]),
    ChatSchema(messages=[user("Goodbye"), assistant("Bye!")]),
]

# Map operator - one-to-one transformation
@map_operator()
def to_uppercase(item):
    for msg in item.messages:
        if isinstance(msg.content, str):
            msg.content = msg.content.upper()
    return item

result = create_dataset(data).execute([
    to_uppercase,
]).collect()
```

--------------------------------

### Dataset API Operations: Filter, Map, Aggregate

Source: https://github.com/azure99/blossomdata/blob/main/src/blossom/dataset/CLAUDE.md

Illustrates a common workflow using the Dataset API, chaining `filter`, `map`, and `aggregate` operations. This example shows how to process data and compute a count.

```python
from blossom.dataset import create_dataset, DatasetEngine
from blossom.operators.aggregate import Count

data = [...] # Your in-memory data

def fn(row):
    # Your filter logic
    return True

def fn2(row):
    # Your map logic
    return row

dataset = create_dataset(data, engine=DatasetEngine.LOCAL)
result = dataset.filter(fn).map(fn2).aggregate(Count())
```

--------------------------------

### Create Custom Transform Operators with Decorators

Source: https://context7.com/azure99/blossomdata/llms.txt

Shows how to create custom batch transformation operators using the `@transform_operator()` decorator. This example duplicates the input data, effectively doubling the dataset size.

```python
from blossom import create_dataset
from blossom.op import transform_operator
from blossom.schema import ChatSchema, user, assistant
import copy

data = [
    ChatSchema(messages=[user("Hello"), assistant("Hi!")]),
    ChatSchema(messages=[user("Goodbye"), assistant("Bye!")]),
]

# Transform operator - batch transformation
@transform_operator()
def duplicate_data(items):
    return items + copy.deepcopy(items)

result = create_dataset(data).execute([
    duplicate_data,
]).collect()
```

--------------------------------

### Chaining Custom Operators for Data Processing

Source: https://context7.com/azure99/blossomdata/llms.txt

Combines multiple custom operators (map, filter, transform, context-map) in a single execution pipeline. This example applies uppercase conversion, filters short messages, duplicates data, and then generates summaries.

```python
from blossom import create_dataset
from blossom.op import map_operator, filter_operator, transform_operator, context_map_operator
from blossom.schema import ChatSchema, ChatRole, TextSchema, user, assistant
import copy

data = [
    ChatSchema(messages=[user("Hello"), assistant("Hi!")]),
    ChatSchema(messages=[user("Goodbye"), assistant("Bye!")]),
]

# Map operator - one-to-one transformation
@map_operator()
def to_uppercase(item):
    for msg in item.messages:
        if isinstance(msg.content, str):
            msg.content = msg.content.upper()
    return item

# Filter operator - keep items matching predicate
@filter_operator()
def filter_short(item):
    return len(item.first_user()) > 5

# Transform operator - batch transformation
@transform_operator()
def duplicate_data(items):
    return items + copy.deepcopy(items)

# Context-aware operator with model access
@context_map_operator(parallel=4)
def generate_summary(context, item):
    prompt = f"Summarize: {item.first_assistant()}"
    summary = context.chat_completion("gpt-4o-mini", [user(prompt)])
    item.metadata["summary"] = summary
    return item

result = create_dataset(data).execute([
    to_uppercase,
    filter_short,
    duplicate_data,
    generate_summary,
]).collect()
```

--------------------------------

### Utilize Context for Configuration and Model Access

Source: https://github.com/azure99/blossomdata/blob/main/README_EN.md

Demonstrates how to create a `Context` object and use it to access configuration settings, retrieve model providers, generate text content using chat completion, and create embedding vectors.

```python
from blossom.context import Context
from blossom.util import user

# Create a context
context = Context()

# Access configuration
config = context.get_config()

# Get model provider
provider = context.get_model("gpt-4o-mini")

# Generate content using a model
response = context.chat_completion(
    "gpt-4o-mini", 
    [user("Hello, how's the weather today?")]
)

# Generate embedding vectors
embedding = context.embedding("text-embedding-3-small", "This is a text")
```

--------------------------------

### Create Dataset with Local Engine

Source: https://github.com/azure99/blossomdata/blob/main/src/blossom/dataset/CLAUDE.md

Demonstrates how to create a dataset using the `create_dataset` function with the `LOCAL` engine. This function takes in-memory data and an optional engine configuration.

```python
from blossom.dataset import create_dataset, DatasetEngine

data = [...] # Your in-memory data

dataset = create_dataset(data, engine=DatasetEngine.LOCAL)
```

--------------------------------

### Dataset Creation and Loading

Source: https://github.com/azure99/blossomdata/blob/main/src/blossom/dataset/AGENTS.md

APIs for creating datasets from in-memory data and loading datasets from files.

```APIDOC
## Dataset Creation and Loading

### Description
Provides functions to initialize a Dataset object either from an in-memory list of schemas or by loading data from specified file paths.

### Methods

#### `create_dataset`

##### Description
Builds a dataset from an in-memory list of `Schema` objects.

##### Parameters

- **data** (`list[Schema]`) - Required - The list of schema objects to create the dataset from.
- **engine** (`DatasetEngine`) - Optional - The engine to use for dataset processing. Defaults to `DatasetEngine.LOCAL`.
- **context** (`Any`) - Optional - An execution context for the dataset.
- **spark_session** (`SparkSession`) - Optional - A SparkSession object, required if using the SPARK engine.

##### Request Example
```python
from blossom.dataset import create_dataset, DatasetEngine
data = [...] # list of Schema objects
dataset = create_dataset(data, engine=DatasetEngine.LOCAL)
```

#### `load_dataset`

##### Description
Loads data from files or directories into a Dataset object. Supports JSON and JSONL formats.

##### Parameters

- **path** (`str` or `list[str]`) - Required - The path(s) to the data file(s) or directory.
- **engine** (`DatasetEngine`) - Optional - The engine to use for dataset processing. Defaults to `DatasetEngine.LOCAL`.
- **data_type** (`DataType`) - Optional - The format of the data. Currently supports `DataType.JSON`. Defaults to `DataType.JSON`.
- **data_handler** (`DataHandler`) - Optional - A custom data handler for processing the loaded data.
- **context** (`Any`) - Optional - An execution context for the dataset.
- **spark_session** (`SparkSession`) - Optional - A SparkSession object, required if using the SPARK engine.

##### Request Example
```python
from blossom.dataset import load_dataset, DatasetEngine, DataType
dataset = load_dataset("/path/to/data", engine=DatasetEngine.LOCAL, data_type=DataType.JSON)
```

### Supported Engines
- `DatasetEngine.LOCAL`
- `DatasetEngine.MULTIPROCESS`
- `DatasetEngine.RAY`
- `DatasetEngine.SPARK`

### Supported Data Types
- `DataType.JSON` (currently the only supported type)
```

--------------------------------

### Create Dataset with Local Engine in BlossomData

Source: https://github.com/azure99/blossomdata/blob/main/README_EN.md

Shows how to create a Dataset object using the `create_dataset` function, specifying the local execution engine. This is a fundamental step for data processing within the framework.

```python
# Create a dataset
dataset = create_dataset(data, engine=DatasetEngine.LOCAL)
```

--------------------------------

### Synthesize Chinese Training Data with BlossomData

Source: https://github.com/azure99/blossomdata/blob/main/README_EN.md

Demonstrates synthesizing verified long-reasoning Chinese training data using math problems and reference answers. It utilizes ChatTranslator, ChatVerifyDistiller, and ChatReasoningContentMerger operators to process and refine the data.

```python
from blossom import *
from blossom.dataframe import *
from blossom.op import *
from blossom.schema import *

# Example math instruction data
data = [
    ChatSchema(
        messages=[
            user("Suppose that $wz = 12-8i$, and $|w| = \sqrt{13}$. What is $|z|$?"),
            assistant("4"),
        ]
    ),
]

# Define operators to use
ops = [
    # Use a non-reasoning model to translate instruction data to Chinese
    ChatTranslator(model="deepseek-chat", target_language="Chinese"),
    # Use a reasoning model to generate answers to questions, and a non-reasoning model to verify correctness
    ChatVerifyDistiller(
        model="deepseek-reasoner",
        mode=ChatVerifyDistiller.Mode.LLM,
        validation_model="deepseek-chat",
    ),
    # Merge reasoning_content into content for training
    ChatReasoningContentMerger(),
]

dataset = create_dataset(data)
result = dataset.execute(ops).collect()
print(result)
```

--------------------------------

### ChatDistiller for Model Responses in Python

Source: https://context7.com/azure99/blossomdata/llms.txt

Shows how to use the ChatDistiller operator to generate model responses for chat data using different distillation strategies (FIRST_TURN, MULTI_TURN, LAST_TURN). It includes parameters for model selection, concurrency, and retries.

```python
from blossom import create_dataset
from blossom.op import ChatDistiller
from blossom.schema import ChatSchema, user, assistant

data = [
    ChatSchema(
        messages=[
            system("You are a helpful assistant."),
            user("Explain machine learning"),
            assistant("placeholder"),  # Will be replaced
            user("Give an example"),
            assistant("placeholder"),  # Will be replaced
        ]
    )
]

# First turn only - generate response for first user message
result = create_dataset(data).execute([
    ChatDistiller(
        model="gpt-4o-mini",
        strategy=ChatDistiller.Strategy.FIRST_TURN,
    )
]).collect()

# Multi-turn - generate responses for all user messages
result = create_dataset(data).execute([
    ChatDistiller(
        model="gpt-4o-mini",
        strategy=ChatDistiller.Strategy.MULTI_TURN,
        parallel=4,  # Process 4 items concurrently
        max_retry=3,
    )
]).collect()

# Last turn only
result = create_dataset(data).execute([
    ChatDistiller(
        model="gpt-4o-mini",
        strategy=ChatDistiller.Strategy.LAST_TURN,
    )
]).collect()
```

--------------------------------

### Interact with LLMs using Context API (Python)

Source: https://context7.com/azure99/blossomdata/llms.txt

Shows how to use the 'Context' object to interact with LLM providers for chat completions and embeddings. It requires the 'blossom' library and a configuration file (e.g., config.yaml). The API supports basic and detailed responses, as well as embedding generation.

```python
from blossom import Context
from blossom.schema import user

# Create context (loads config from config.yaml)
context = Context()

# Get configuration
config = context.get_config()

# Simple chat completion
response = context.chat_completion(
    "gpt-4o-mini",
    [user("What is 2 + 2?")]
)
print(response)  # "4"

# Chat completion with detailed response
detailed_response = context.chat_completion_with_details(
    "gpt-4o-mini",
    [user("Explain gravity")],
    extra_params={"temperature": 0.7, "max_tokens": 500}
)
print(detailed_response.choices[0].message.content)
print(detailed_response.usage.total_tokens)

# Generate embeddings
embedding = context.embedding(
    "text-embedding-3-small",
    "This is a sample text"
)
print(len(embedding))  # Embedding dimension (e.g., 1536)

# Embedding with details
embedding_response = context.embedding_with_details(
    "text-embedding-3-small",
    "Another sample"
)
print(embedding_response.usage.total_tokens)

```

--------------------------------

### Ray DataFrame: Initialize Ray if Needed

Source: https://github.com/azure99/blossomdata/blob/main/src/blossom/dataset/CLAUDE.md

Shows how RayDataFrame automatically initializes a Ray session if one is not already active. This is crucial for distributed operations with Ray.

```python
import ray
from blossom.dataset import create_dataset, DatasetEngine

# ray.init() is called automatically by RayDataFrame if not already initialized
data = [...] # Your in-memory data

dataset = create_dataset(data, engine=DatasetEngine.RAY)
```

--------------------------------

### RowSchema for Structured Data in Python

Source: https://context7.com/azure99/blossomdata/llms.txt

Illustrates the usage of RowSchema for representing structured key-value data. It shows how to initialize a RowSchema object and access its data using both dictionary-like syntax and attribute access.

```python
from blossom.schema import RowSchema

row = RowSchema(
    data={
        "name": "John",
        "age": 30,
        "scores": [85, 90, 92],
        "active": True
    }
)

# Access data using dict-like syntax
print(row["name"])  # "John"
print(row["age"])   # 30

# Access via data attribute
print(row.data["scores"])  # [85, 90, 92]
```

--------------------------------

### Load and Process Dataset with Ray Engine

Source: https://github.com/azure99/blossomdata/blob/main/README_EN.md

Loads a dataset from a JSONL file using Ray as the execution engine. It then applies a series of transformations including repartitioning, filtering by language, sorting by feedback score, adding metadata for context length, limiting the number of data points, and executing chat translation and distillation operators. Finally, it caches the processed dataset and demonstrates collecting results and writing to a JSONL file.

```python
from blossom.engine import DatasetEngine
from blossom.handler import DefaultDataHandler
from blossom.op import ChatTranslator, ChatDistiller

# Load dataset using Ray engine
dataset = load_dataset(
    "example/data/chat.jsonl",
    engine=DatasetEngine.RAY,
    data_handler=DefaultDataHandler(),
)

# Data transformation pipeline
dataset = (
    dataset.repartition(2)
    .filter(lambda x: x.metadata["language"] == "en")
    .sort(lambda x: x.metadata["feedback"], ascending=False)
    .add_metadata(
        lambda x: {
            "context_length": sum(len(message.content) for message in x.messages)
        }
    )
    .limit(4)
    .execute(
        [
            ChatTranslator(model="gpt-4o-mini", target_language="Chinese"),
            ChatDistiller(model="gpt-4o-mini", parallel=4),
        ]
    )
    .cache()
)

# Collect and write results
results = dataset.collect()
dataset.write_json("output.jsonl")

# Aggregation operations
agg_result = dataset.aggregate(
    Sum(lambda x: x.metadata["context_length"]),
    Mean(lambda x: x.metadata["context_length"]),
    Count(),
)

# Group and aggregate
grouped_count = dataset.group_by(lambda x: x.metadata["country"]).count().collect()

print(results)
print(agg_result)
print(grouped_count)
```

--------------------------------

### DataHandler and Schema Conventions

Source: https://github.com/azure99/blossomdata/blob/main/src/blossom/dataset/AGENTS.md

Information on DataHandlers and Schema conventions used for data processing.

```APIDOC
## DataHandler and Schema Conventions

### Description
Explains the roles of `DataHandler` and `Schema` in processing and structuring data within the Blossom framework.

### DataHandlers

- **`DefaultDataHandler`**: 
  - If a `type` field exists in the schema, it uses `Schema.from_dict` for instantiation.
  - Otherwise, it wraps dictionaries in `RowSchema`.

- **`DictDataHandler(preserve_metadata=False)`**: 
  - Designed for workflows involving raw dictionaries.
  - The `preserve_metadata` flag controls whether metadata is retained during processing.

### Schema Conventions

- **Metadata**: The `Schema` object supports metadata updates via `add_metadata` and `drop_metadata` methods on the `Dataset` object.

### Aggregations

- **Base Classes**: Aggregations are built upon `AggregateFunc` and `RowAggregateFunc`.
- **Multiple Aggregations**: When multiple aggregations are performed, the result is returned as a dictionary where keys correspond to the `agg.name` attribute of each aggregation function.
```

--------------------------------

### Create Custom Operators with Class Inheritance (Python)

Source: https://context7.com/azure99/blossomdata/llms.txt

Demonstrates creating custom operators like 'SelfQAGenerator' and 'QualityFilter' by inheriting from base classes (MapOperator, FilterOperator). This allows for complex data processing pipelines by extending existing functionalities. It requires the 'blossom' library.

```python
from blossom import create_dataset
from blossom.op import MapOperator, FilterOperator, TransformOperator
from blossom.schema import ChatSchema, TextSchema, user, assistant
from blossom.util import loads_markdown_first_json

class SelfQAGenerator(MapOperator):
    """Generates Q&A pairs from text content using LLM."""

    def __init__(self, model: str = "gpt-4o-mini", parallel: int = 1):
        super().__init__(parallel=parallel)
        self.model = model

    def process_item(self, item):
        prompt = (
            "Based on the given text, generate a question and detailed answer."
            "\nOutput JSON with 'question' and 'answer' string fields only."
            f"\nText: {item.content}"
        )
        raw_result = self.context.chat_completion(self.model, [user(prompt)])
        result = loads_markdown_first_json(raw_result)
        return ChatSchema(
            messages=[
                user(result["question"]),
                assistant(result["answer"]),
            ],
            metadata=item.metadata,
        )

class QualityFilter(FilterOperator):
    """Filters items based on quality score threshold."""

    def __init__(self, min_score: float = 0.7, parallel: int = 1):
        super().__init__(parallel=parallel)
        self.min_score = min_score

    def process_item(self, item) -> bool:
        score = item.metadata.get("quality_score", 0)
        return score >= self.min_score

# Usage
data = [TextSchema(content="Python is a versatile programming language.")]
result = create_dataset(data).execute([
    SelfQAGenerator(model="gpt-4o-mini"),
]).collect()

```

--------------------------------

### Create Context-Aware Map Operators with Decorators

Source: https://context7.com/azure99/blossomdata/llms.txt

Demonstrates creating context-aware map operators using `@context_map_operator()`. This operator can access external context (like a model) to perform operations, such as summarizing assistant messages, with parallel processing.

```python
from blossom import create_dataset
from blossom.op import context_map_operator
from blossom.schema import ChatSchema, user, assistant

data = [
    ChatSchema(messages=[user("Hello"), assistant("Hi!")]),
    ChatSchema(messages=[user("Goodbye"), assistant("Bye!")]),
]

# Context-aware operator with model access
@context_map_operator(parallel=4)
def generate_summary(context, item):
    prompt = f"Summarize: {item.first_assistant()}"
    summary = context.chat_completion("gpt-4o-mini", [user(prompt)])
    item.metadata["summary"] = summary
    return item

result = create_dataset(data).execute([
    generate_summary,
]).collect()
```

--------------------------------

### Minimal Skeleton for a Text Operator (Python)

Source: https://github.com/azure99/blossomdata/blob/main/src/blossom/op/AGENTS.md

Provides a basic structure for creating a text-based operator using the MapOperator base class. It demonstrates casting the input item to a text schema, performing mutations on content or metadata, and casting it back to a base schema.

```python
from blossom.op.map_operator import MapOperator
from blossom.schema.schema import Schema

class MyTextOp(MapOperator):
    def process_item(self, item: Schema) -> Schema:
        _item = self._cast_text(item)
        # mutate _item.content or _item.metadata here
        return self._cast_base(_item)
```

--------------------------------

### Load JSON Dataset from Files

Source: https://github.com/azure99/blossomdata/blob/main/src/blossom/dataset/CLAUDE.md

Shows how to load a dataset from JSON files or directories using the `load_dataset` function. It supports various engines and data types, defaulting to JSON.

```python
from blossom.dataset import load_dataset, DatasetEngine, DataType

file_path = "/path/to/your/data.jsonl"
dataset = load_dataset(file_path, engine=DatasetEngine.LOCAL, data_type=DataType.JSON)
```

--------------------------------

### Create Dataset from In-Memory Data (Python)

Source: https://context7.com/azure99/blossomdata/llms.txt

Creates a BlossomData Dataset from a list of Schema objects in memory. Supports local, Ray, and Spark execution engines for distributed processing.

```python
from blossom import create_dataset, DatasetEngine
from blossom.schema import ChatSchema, user, assistant

# Create data in memory
data = [
    ChatSchema(
        messages=[
            user("What is 2 + 2?"),
            assistant("4"),
        ],
        metadata={"difficulty": "easy"}
    ),
    ChatSchema(
        messages=[
            user("Explain quantum computing"),
            assistant("Quantum computing uses quantum bits..."),
        ],
        metadata={"difficulty": "hard"}
    ),
]

# Create dataset with local engine (default)
dataset = create_dataset(data)

# Create dataset with Ray distributed engine
dataset = create_dataset(data, engine=DatasetEngine.RAY)

# Create dataset with Spark engine
from pyspark.sql import SparkSession
spark = SparkSession.builder.getOrCreate()
dataset = create_dataset(data, engine=DatasetEngine.SPARK, spark_session=spark)
```

--------------------------------

### DataFrame Engines and Behavior

Source: https://github.com/azure99/blossomdata/blob/main/src/blossom/dataset/AGENTS.md

Details on the different DataFrame engines and their specific behaviors.

```APIDOC
## DataFrame Engines and Behavior

### Description
This section describes the characteristics and behavior of the different DataFrame engines supported by Blossom, including `LocalDataFrame`, `MultiProcessDataFrame`, `RayDataFrame`, and `SparkDataFrame`.

### Engine Details

#### `LocalDataFrame`

- **Behavior**: Operates on in-memory lists.
- **`repartition`**: This operation is a no-op and issues a warning, as the data is already in memory.
- **`read_json`**: Can read `.json` or `.jsonl` files from files or directories, processing them line by line.

#### `MultiProcessDataFrame`

- **Behavior**: Utilizes `ProcessPoolExecutor` and `cloudpickle` for distributed execution of `map`, `filter`, and `transform` operations across multiple processes.
- **Function Requirements**: Functions used with this engine must be picklable. Be cautious with shared state, as it is not safe across processes.
- **Aggregations/Grouping**: These operations are executed in-process.

#### `RayDataFrame`

- **Behavior**: Wraps Ray Datasets. Converts `Schema` objects to row dictionaries.
- **Initialization**: Automatically calls `ray.init()` if a Ray session is not already active.
- **`write_json`**: Uses a JSONL datasink for writing output.

#### `SparkDataFrame`

- **Behavior**: Wraps Spark RDDs.
- **`read_json`**: Expects line-delimited JSON text files.
- **`write_json`**: Writes output to a Spark text output directory.
- **Requirement**: Requires an active `SparkSession` to be provided.
```

--------------------------------

### Dataset I/O: Write and Read JSON/Parquet (Python)

Source: https://context7.com/azure99/blossomdata/llms.txt

Handles reading and writing datasets to and from JSON Lines (.jsonl) and Parquet (.parquet) file formats. This enables data persistence and interoperability. It uses `write_json`, `write_parquet`, and `load_dataset` methods.

```python
from blossom import create_dataset, load_dataset
from blossom.schema import ChatSchema, user, assistant

data = [
    ChatSchema(messages=[user("Hello"), assistant("Hi")])
]

# Write to JSONL file
create_dataset(data).write_json("output.jsonl")

# Write to Parquet file
create_dataset(data).write_parquet("output.parquet")

# Read from file
loaded = load_dataset("output.jsonl").collect()

# Chain operations and write
(
    load_dataset("input.jsonl")
    .filter(lambda x: len(x.messages) > 0)
    .shuffle()
    .limit(1000)
    .write_json("filtered_output.jsonl")
)
```

--------------------------------

### Dataset Transformations and Operations

Source: https://github.com/azure99/blossomdata/blob/main/src/blossom/dataset/AGENTS.md

Core API for manipulating Dataset objects through a chainable interface.

```APIDOC
## Dataset Transformations and Operations

### Description
This section details the chainable API for transforming and operating on `Dataset` objects, providing methods for filtering, mapping, sorting, and more.

### Core API Methods

- **`map(fn)`**: Applies a function `fn` to each element of the dataset.
- **`filter(fn)`**: Filters elements based on a predicate function `fn`.
- **`transform(fn)`**: Applies a transformation function `fn` to the dataset.
- **`sort(key_fn, ascending=True)`**: Sorts the dataset based on a key function.
- **`limit(n)`**: Limits the dataset to the first `n` elements.
- **`shuffle()`**: Shuffles the dataset elements.
- **`repartition(n)`**: Repartitions the dataset into `n` partitions (behavior may vary by engine).
- **`split(ratios)`**: Splits the dataset into multiple subsets based on given ratios.
- **`union(other_dataset)`**: Combines the current dataset with another dataset.
- **`cache()`**: Caches the dataset in memory for faster access.
- **`collect()`**: Collects all elements of the dataset into a list.

### Execution and Metadata

- **`execute(operators)`**: Executes a list of operators on the dataset after initializing the context.
- **`add_metadata(key, value)`**: Adds metadata to the dataset's schema.
- **`drop_metadata(key)`**: Removes metadata associated with a key from the schema.

### I/O Operations

- **`read_json(path, ...)`**: Reads JSON/JSONL data from files or directories (specific implementation depends on the engine).
- **`write_json(path, ...)`**: Writes the dataset content as JSONL to the specified path (specific implementation depends on the engine).

### Aggregations

- **`aggregate(*AggregateFunc)`**: Performs aggregations on the dataset. Returns a dictionary keyed by aggregation name.
- **`sum(*columns)`**: Calculates the sum of specified columns.
- **`mean(*columns)`**: Calculates the mean of specified columns.
- **`count()`**: Counts the number of elements in the dataset.
- **`min(*columns)`**: Finds the minimum value in specified columns.
- **`max(*columns)`**: Finds the maximum value in specified columns.
- **`variance(*columns)`**: Calculates the variance of specified columns.
- **`stddev(*columns)`**: Calculates the standard deviation of specified columns.
- **`unique(*columns)`**: Finds the unique values in specified columns.

### Grouping

- **`group_by(func, name="group")`**: Groups the dataset based on a function `func` and returns a `GroupedDataset` object. The `GroupedDataset` provides the same aggregation helpers (`sum`, `mean`, etc.).

### Example Usage
```python
from blossom.dataset import create_dataset, DatasetEngine
from blossom.aggregations import Count

data = [...] # list of Schema objects

dataset = create_dataset(data, engine=DatasetEngine.LOCAL)

# Example transformation and aggregation
filtered_dataset = dataset.filter(lambda x: x.value > 10)
mapped_dataset = filtered_dataset.map(lambda x: x.value * 2)
result = mapped_dataset.aggregate(Count(), sum=Sum("value"))

# Example grouping and aggregation
grouped_data = dataset.group_by(lambda x: x.category).mean("value")
```
```

--------------------------------

### Load Dataset from Files (Python)

Source: https://context7.com/azure99/blossomdata/llms.txt

Loads a BlossomData Dataset from JSON or Parquet files. Supports loading from single files or lists of files, with options for different engines and data types.

```python
from blossom import load_dataset, DatasetEngine, DataType
from blossom.dataframe import DefaultDataHandler

# Load from JSONL file with local engine
dataset = load_dataset("data/chat.jsonl")

# Load from multiple files with Ray engine
dataset = load_dataset(
    ["data/part1.jsonl", "data/part2.jsonl"],
    engine=DatasetEngine.RAY,
    data_type=DataType.JSON,
    data_handler=DefaultDataHandler(),
)

# Load from Parquet file
dataset = load_dataset(
    "data/training_data.parquet",
    data_type=DataType.PARQUET,
)
```

--------------------------------

### Sort Dataset by Key (Python)

Source: https://context7.com/azure99/blossomdata/llms.txt

Sorts the dataset rows based on a specified key function in either ascending or descending order. This is essential for ordering data for analysis or presentation. The `sort` method takes a key function and an `ascending` boolean parameter.

```python
from blossom import create_dataset
from blossom.schema import RowSchema

data = [
    RowSchema(data={"name": "Alice", "score": 85}),
    RowSchema(data={"name": "Bob", "score": 92}),
    RowSchema(data={"name": "Charlie", "score": 78}),
]

# Sort by score descending
sorted_data = (
    create_dataset(data)
    .sort(lambda x: x["score"], ascending=False)
    .collect()
)
for item in sorted_data:
    print(f"{item['name']}: {item['score']}")
# Output: Bob: 92, Alice: 85, Charlie: 78
```

--------------------------------

### Create and Use TextSchema (Python)

Source: https://context7.com/azure99/blossomdata/llms.txt

Defines and utilizes the `TextSchema` for representing plain text content, suitable for text processing tasks. It allows storing the text content and associated metadata, such as language or word count.

```python
from blossom.schema import TextSchema

text = TextSchema(
    content="This is a sample text for processing.",
    metadata={"language": "en", "word_count": 7}
)

# Access content
print(text.content)
```

--------------------------------

### ChatVerifyDistiller for Response Validation in Python

Source: https://context7.com/azure99/blossomdata/llms.txt

Explains the ChatVerifyDistiller operator for generating and validating model responses against references using regex, LLM judges, or custom functions. It's suitable for tasks like math problems or any scenario requiring answer verification.

```python
from blossom import create_dataset
from blossom.op import ChatVerifyDistiller
from blossom.schema import ChatSchema, user, assistant

# Math problems with reference answers
data = [
    ChatSchema(
        messages=[
            user("What is 15 * 7?"),
            assistant("105"),  # Reference answer
        ]
    )
]

# Validate using regex (extracts last number)
result = create_dataset(data).execute([
    ChatVerifyDistiller(
        model="gpt-4o-mini",
        mode=ChatVerifyDistiller.Mode.REGEX,
        max_retry=3,
    )
]).collect()

# Validate using LLM judge
result = create_dataset(data).execute([
    ChatVerifyDistiller(
        model="deepseek-reasoner",
        mode=ChatVerifyDistiller.Mode.LLM,
        validation_model="gpt-4o-mini",
        max_retry=2,
    )
]).collect()

# Custom validation function
def custom_validator(question, reference, model_answer):
    return reference.lower() in model_answer.lower()

result = create_dataset(data).execute([
    ChatVerifyDistiller(
        model="gpt-4o-mini",
        mode=ChatVerifyDistiller.Mode.FUNCTION,
        validation_function=custom_validator,
    )
]).collect()
```

--------------------------------

### Create and Use ChatSchema (Python)

Source: https://context7.com/azure99/blossomdata/llms.txt

Defines and utilizes the `ChatSchema` for representing conversational data, including system, user, and assistant messages, along with optional metadata. It provides helper functions for message creation and methods for accessing, modifying, and converting chat data.

```python
from blossom.schema import ChatSchema, ChatMessage, user, assistant, system

# Create using helper functions
chat = ChatSchema(
    messages=[
        system("You are a helpful assistant."),
        user("What is Python?"),
        assistant("Python is a programming language."),
    ],
    metadata={"source": "manual", "quality": "high"}
)

# Access message content
first_user_msg = chat.first_user()  # "What is Python?"
last_assistant_msg = chat.last_assistant()  # "Python is a programming language."

# Modify messages
chat.add_user("Can you give an example?")
chat.add_assistant("Here's a simple example: print('Hello')")
chat.remove_last_assistant()

# Convert to dict
data = chat.model_dump()
```

--------------------------------

### Mark Data as Failed in Python

Source: https://context7.com/azure99/blossomdata/llms.txt

Demonstrates how to mark a text object as failed and access its failure status and reason. This is useful for error handling during data processing pipelines.

```python
text.mark_failed("Processing error: invalid encoding")
print(text.failed)  # True
print(text.failure_reason)  # "Processing error: invalid encoding"
```

--------------------------------

### Implement Custom Map Operator by Inheriting from Operator

Source: https://github.com/azure99/blossomdata/blob/main/README_EN.md

Shows how to implement a custom map operator, `SelfQA`, by inheriting from the `MapOperator` class. The `process_item` method is overridden to generate a question-answer pair based on the input text using a chat completion model.

```python
from blossom.op import MapOperator
from blossom.schema import ChatSchema
from blossom.util import loads_markdown_first_json, user, assistant

class SelfQA(MapOperator):
    def process_item(self, item):
        self_qa_prompt = (
            "Based on the given text, freely generate a question and a corresponding long answer.\n"
            "Your output should be a JSON with two string fields: question and answer, without any other irrelevant explanations.\n"
            f"Given text: {item.content}"
        )
        raw_result = self.context.chat_completion("gpt-4o-mini", [user(self_qa_prompt)])
        result = loads_markdown_first_json(raw_result)
        return ChatSchema(
            messages=[
                user(result["question"]),
                assistant(result["answer"]),
            ]
        )
```

--------------------------------

### Transform Dataset by Duplicating Items (Python)

Source: https://context7.com/azure99/blossomdata/llms.txt

Applies a batch transformation to dataset items by duplicating each item and assigning new IDs. The `transform` method operates on lists, enabling efficient batch operations. It requires the `copy` module for deep copying and `create_dataset`.

```python
import copy
from blossom import create_dataset
from blossom.schema import ChatSchema, user, assistant

data = [
    ChatSchema(messages=[user("Hello"), assistant("Hi")])
]

# Duplicate data in batch
def duplicate_items(items):
    duplicated = items + copy.deepcopy(items)
    for idx, item in enumerate(duplicated):
        item.id = str(idx)
    return duplicated

result = create_dataset(data).transform(duplicate_items).collect()
print(len(result))  # 2
```

--------------------------------

### ChatTranslator for Chat Message Translation in Python

Source: https://context7.com/azure99/blossomdata/llms.txt

Demonstrates the ChatTranslator operator for translating chat messages to a target language. It supports translating all messages, specific roles (e.g., user messages only), and includes reasoning content.

```python
from blossom import create_dataset
from blossom.op import ChatTranslator
from blossom.schema import ChatSchema, ChatRole, user, assistant

data = [
    ChatSchema(messages=[
        user("Hello, how are you?"),
        assistant("I'm doing well, thank you!"),
    ])
]

# Translate all messages to Chinese
result = create_dataset(data).execute([
    ChatTranslator(
        model="gpt-4o-mini",
        target_language="Chinese",
        parallel=2,
    )
]).collect()

# Translate only user messages
result = create_dataset(data).execute([
    ChatTranslator(
        model="gpt-4o-mini",
        target_language="Japanese",
        roles=[ChatRole.USER],  # Only translate user messages
    )
]).collect()

# Translate including reasoning content
result = create_dataset(data).execute([
    ChatTranslator(
        model="gpt-4o-mini",
        target_language="Spanish",
        translate_reasoning=True,
    )
]).collect()
```

--------------------------------

### Configure Model Providers

Source: https://context7.com/azure99/blossomdata/llms.txt

Defines model provider settings in a YAML configuration file. This allows BlossomData to connect to different LLM services like OpenAI.

```yaml
models:
- name: gpt-4o
  provider: openai
  config:
    base_url: https://api.openai.com/v1
    key: sk-xxx
- name: deepseek-chat
  provider: openai
  config:
    base_url: https://api.deepseek.com/v1
    key: sk-xxx
```

--------------------------------

### Execute Operators on Dataset (Python)

Source: https://context7.com/azure99/blossomdata/llms.txt

Executes a sequence of operators on a BlossomData Dataset. This is the primary method for applying data transformations and processing pipelines.

```python
from blossom import create_dataset
from blossom.op import ChatTranslator, ChatDistiller, ChatLengthFilter
from blossom.schema import ChatSchema, user, assistant

data = [
    ChatSchema(messages=[user("Hello"), assistant("Hi there!")])
]

# Execute multiple operators in sequence
result = (
    create_dataset(data)
    .execute([
        ChatTranslator(model="gpt-4o-mini", target_language="Chinese"),
        ChatDistiller(model="gpt-4o-mini"),
        ChatLengthFilter(total_max_len=8192),
    ])
    .collect()
)
print(result[0].messages)  # Translated and distilled messages
```

--------------------------------

### Group Dataset by Key and Aggregate (Python)

Source: https://context7.com/azure99/blossomdata/llms.txt

Groups dataset rows by a specified key function (e.g., 'country') to perform grouped aggregations. This enables calculating statistics for distinct groups within the data. It utilizes `group_by` followed by aggregation methods like `count` or `aggregate`.

```python
from blossom import create_dataset
from blossom.dataframe import Count, Sum, Mean
from blossom.schema import RowSchema
import random

data = [
    RowSchema(data={"country": random.choice(["US", "CN", "JP"]), "score": random.randint(1, 100)})
    for _ in range(100)
]

# Group by country and count
country_counts = (
    create_dataset(data)
    .group_by(lambda x: x["country"], name="country")
    .count()
    .collect()
)
for row in country_counts:
    print(f"{row['country']}: {row['count']}")

# Group by country with multiple aggregations
stats = (
    create_dataset(data)
    .group_by(lambda x: x["country"])
    .aggregate(
        Count(),
        Sum(lambda x: x["score"]),
        Mean(lambda x: x["score"]),
    )
    .collect()
)
```

--------------------------------

### Translate Text Content with TextTranslator

Source: https://context7.com/azure99/blossomdata/llms.txt

Translates plain text content to a target language using a specified model. It allows for parallel processing and retries in case of translation failures. Useful for multilingual data processing.

```python
from blossom import create_dataset
from blossom.op import TextTranslator
from blossom.schema import TextSchema

data = [
    TextSchema(content="Hello, this is a sample text."),
    TextSchema(content="Machine learning is fascinating."),
]

result = create_dataset(data).execute([
    TextTranslator(
        model="gpt-4o-mini",
        target_language="French",
        parallel=2,
        max_retry=3,
    )
]).collect()
for item in result:
    print(item.content)
```

--------------------------------

### Map Function over Dataset (Python)

Source: https://context7.com/azure99/blossomdata/llms.txt

Applies a one-to-one transformation function to each item in a BlossomData Dataset. Useful for converting data formats or extracting specific information.

```python
from blossom import create_dataset
from blossom.schema import ChatSchema, TextSchema, user, assistant

data = [
    ChatSchema(
        messages=[user("Question"), assistant("Answer")],
        metadata={"source": "qa_dataset"}
    )
]

# Transform chat data to text format
result = (
    create_dataset(data)
    .map(lambda x: TextSchema(
        id=x.id,
        content=f"Q: {x.messages[0].content}\nA: {x.messages[1].content}",
        metadata=x.metadata,
    ))
    .collect()
)
print(result[0].content)
# Output: Q: Question
#         A: Answer
```

--------------------------------

### Aggregate Dataset with Built-in Functions (Python)

Source: https://context7.com/azure99/blossomdata/llms.txt

Performs aggregation operations on a dataset using predefined functions like Sum, Mean, Count, Min, Max, Variance, and StdDev. This allows for calculating statistics across the dataset or specific columns. It requires importing aggregation functions from `blossom.dataframe`.

```python
from blossom import create_dataset
from blossom.dataframe import Sum, Mean, Count, Min, Max, Variance, StdDev
from blossom.schema import RowSchema
import random

data = [
    RowSchema(data={"score": random.randint(1, 100), "country": random.choice(["US", "CN"])}
    for _ in range(100)
]

dataset = create_dataset(data)

# Single aggregation
total_count = dataset.count()
avg_score = dataset.mean(lambda x: x["score"])
min_score = dataset.min(lambda x: x["score"])
max_score = dataset.max(lambda x: x["score"])

# Multiple aggregations at once
result = dataset.aggregate(
    Count(),
    Sum(lambda x: x["score"]),
    Mean(lambda x: x["score"]),
    Min(lambda x: x["score"]),
    Max(lambda x: x["score"]),
)
print(result)  # {'count': 100, 'sum': 5050, 'mean': 50.5, 'min': 1, 'max': 100}
```

--------------------------------

### Filter Dataset by Content Length (Python)

Source: https://context7.com/azure99/blossomdata/llms.txt

Filters a dataset to include only items where the first user message has a length greater than 10 characters. This is useful for identifying longer prompts in conversational data. It relies on the `create_dataset` function and a lambda for filtering.

```python
from blossom import create_dataset

data = [
    ChatSchema(messages=[user("Hello"), assistant("Hi")])
]

long_prompts = create_dataset(data).filter(
    lambda x: len(x.first_user()) > 10
).collect()
print(len(long_prompts))  # 1
```

--------------------------------

### Spark DataFrame JSON Input/Output

Source: https://github.com/azure99/blossomdata/blob/main/src/blossom/dataset/AGENTS.md

Details how to read and write JSON data using Spark DataFrames within the Blossomdata framework. This requires a SparkSession and assumes line-delimited JSON text.

```python
from pyspark.sql import SparkSession
from blossom.dataset import load_dataset, DatasetEngine, DataType

spark_session = SparkSession.builder.appName("BlossomSpark").getOrCreate()
input_path = "/path/to/spark/input.jsonl"
output_path = "/path/to/spark/output"

# Reading JSONL with Spark
dataset_spark = load_dataset(input_path, engine=DatasetEngine.SPARK, data_type=DataType.JSON, spark_session=spark_session)

# Writing JSONL with Spark (assuming dataset_spark has a write_json method)
dataset_spark.write_json(output_path)

```

--------------------------------

### ChatContentFilter for Filtering Chat Data in Python

Source: https://context7.com/azure99/blossomdata/llms.txt

Demonstrates the ChatContentFilter operator, which filters chat data based on content matching specified patterns. This is useful for removing unwanted or sensitive information from chat logs.

```python
from blossom import create_dataset
from blossom.op import ChatContentFilter
from blossom.schema import ChatSchema, ChatRole, user, assistant

data = [
    ChatSchema(messages=[user("Hello"), assistant("Hi there!")]),
    ChatSchema(messages=[user("Hello"), assistant("I cannot help with that.")]),
    ChatSchema(messages=[user("Hello"), assistant("As an AI, I don't have opinions.")]),
]
```

--------------------------------

### Define Custom Map, Filter, and Transform Operators

Source: https://github.com/azure99/blossomdata/blob/main/README_EN.md

Demonstrates how to define custom operators using decorators. `uppercase_text` is a map operator that converts text content to uppercase. `filter_short_text` is a filter operator that keeps text longer than 10 characters. `batch_process` is a transform operator that filters items containing a specific keyword.

```python
from blossom.op import map_operator, filter_operator, transform_operator

@map_operator()
def uppercase_text(item):
    item.content = item.content.upper()
    return item

@filter_operator()
def filter_short_text(item):
    return len(item.content) > 10

@transform_operator()
def batch_process(items):
    return [item for item in items if "keyword" in item.content]
```