### Python Doctest Example with Setup

Source: https://github.com/mosaicml/streaming/blob/main/STYLE_GUIDE.md

An example of a Python function docstring that includes a `.. testsetup::` block for optional setup code and a `.. testcode::` block for the actual executable example.

```python
import torch
from typing import Optional

def my_function(x: Optional[torch.Tensor]) -> torch.Tensor:
    """blah function

    Args:
        input (torch.Tensor): Your guess.

    Returns:
        torch.Tensor: How good your input is.

    Raises:
        ValueError: If your input is negative.

    Example:
        .. testsetup::

            # optional setup section, not shown in docs
            import torch
            x = torch.randn(42)


        .. testcode::

            # shown in docs; runs after testsetup
            my_function(x)
    """
    ...
```

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

### Install and Install Pre-Commit Hooks

Source: https://github.com/mosaicml/streaming/blob/main/STYLE_GUIDE.md

Install the project with development dependencies and set up pre-commit hooks to enforce style checks before each commit.

```bash
pip install '.[dev]'
pre-commit install
```

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

### Install Development Dependencies

Source: https://github.com/mosaicml/streaming/blob/main/CONTRIBUTING.md

Installs the necessary dependencies for testing and linting the code. Use the optional flag to install all dependencies.

```bash
pip install -e '.[dev]'
```

```bash
pip install -e '.[all]'
```

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

### Install Dependencies for Streaming and Training

Source: https://github.com/mosaicml/streaming/blob/main/examples/facesynthetics.ipynb

Installs necessary packages including mmsegmentation, mmcv, mosaicml, and mosaicml-streaming. Optional AWS CLI installation is also included.

```python
%pip install mmsegmentation mmcv mmcv-full

%pip install mosaicml
# To install from source instead of the last release, comment the command above and uncomment the following one.
# %pip install git+https://github.com/mosaicml/composer.git

%pip install mosaicml-streaming
# To install from source instead of the last release, comment the command above and uncomment the following one.
# %pip install git+https://github.com/mosaicml/streaming.git

# (Optional) To upload a streaming dataset to an AWS S3 bucket
%pip install awscli
```

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

### Install Required Libraries

Source: https://github.com/mosaicml/streaming/blob/main/examples/spark_dataframe_to_MDS.ipynb

Installs or upgrades fsspec, datasets, and transformers. Use this to set up your environment for the tutorial.

```python
%pip install --upgrade fsspec  datasets transformers
```

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

### Install Simulator Dependencies

Source: https://github.com/mosaicml/streaming/blob/main/docs/source/distributed_training/performance_tuning.md

Install the mosaicml-streaming library with simulator-specific dependencies. This is a prerequisite for running the simulator.

```bash
pip install --upgrade "mosaicml-streaming[simulator]"
```

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

### Install TensorFlow

Source: https://github.com/mosaicml/streaming/blob/main/streaming/text/convert/enwiki/mds/README.md

Install TensorFlow using pip. This is a prerequisite for running the dataset creation scripts.

```bash
pip3 install tensorflow
```

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

### Install Required Libraries

Source: https://github.com/mosaicml/streaming/blob/main/docs/source/preparing_datasets/spark_dataframe_to_mds.ipynb

Install necessary packages including fsspec, datasets, transformers, and mosaicml-streaming.

```python
%pip install --upgrade fsspec  datasets transformers
```

```python
%pip install mosaicml-streaming
```

```python
%pip install pyspark==3.4.1
```

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

### Install mosaicml-streaming

Source: https://github.com/mosaicml/streaming/blob/main/docs/source/how_to_guides/cifar10.ipynb

Installs the mosaicml-streaming package and its dependencies. Optionally, install from source or install AWS CLI for S3 uploads.

```bash
%pip install mosaicml-streaming
# To install from source instead of the last release, comment the command above and uncomment the following one.
# %pip install git+https://github.com/mosaicml/streaming.git

# (Optional) To upload a streaming dataset to an AWS S3 bucket
%pip install awscli
```

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

### Install mosaicml-streaming

Source: https://github.com/mosaicml/streaming/blob/main/docs/source/index.md

Install the mosaicml-streaming library using pip. This is a prerequisite for using StreamingDataset.

```bash
pip install mosaicml-streaming
```

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

### AWS S3 Config File Example

Source: https://github.com/mosaicml/streaming/blob/main/docs/source/how_to_guides/configure_cloud_storage_credentials.md

Example of the AWS config file structure. Ensure the region and output format are correctly specified.

```ini
[default]
region=<your region, e.g. us-west-2>
output=json
```

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

### Install mosaicml-streaming

Source: https://github.com/mosaicml/streaming/blob/main/docs/source/preparing_datasets/parallel_dataset_conversion.ipynb

Install the mosaicml-streaming package. This is a prerequisite for using MDSWriter and StreamingDataset.

```python
%pip install mosaicml-streaming
```

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

### Verify Streaming Installation

Source: https://github.com/mosaicml/streaming/blob/main/docs/source/index.md

Verify the installation of the mosaicml-streaming library by checking its version. This command can be run directly from the command line.

```python
python -c "import streaming; print(streaming.__version__)"
```

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

### Cleanup Tutorial Files

Source: https://github.com/mosaicml/streaming/blob/main/examples/synthetic_nlp.ipynb

Remove temporary files and directories created during the tutorial to free up disk space. This is a standard cleanup procedure after running examples.

```python
shutil.rmtree(out_root, ignore_errors=True)
shutil.rmtree(local, ignore_errors=True)
```

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

### AWS S3 Credentials File Example

Source: https://github.com/mosaicml/streaming/blob/main/docs/source/how_to_guides/configure_cloud_storage_credentials.md

Example of the AWS credentials file structure. Use your AWS access key ID and secret access key.

```ini
[default]
aws_access_key_id=<key ID>
aws_secret_access_key=<application key>
```

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

### Install PySpark

Source: https://github.com/mosaicml/streaming/blob/main/examples/spark_dataframe_to_MDS.ipynb

Installs a specific version of PySpark. Ensure compatibility with your Spark environment.

```python
%pip install pyspark==3.4.1
```

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

### Install Pre-commit Hooks

Source: https://github.com/mosaicml/streaming/blob/main/CONTRIBUTING.md

Configures pre-commit to automatically format code before each commit. This ensures code style consistency.

```bash
pre-commit install
```

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

### Specify Compression Algorithm

Source: https://github.com/mosaicml/streaming/blob/main/docs/source/preparing_datasets/basic_dataset_conversion.md

Choose a compression algorithm to reduce shard file sizes and egress costs. Examples show specifying the algorithm name and optionally a compression level.

```python
compression = 'zstd' # zstd, defaults to level 3.
```

```python
compression = 'zstd:9' # zstd, specifying level 9.
```

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

### Instantiate StreamingDataset and Access Size

Source: https://github.com/mosaicml/streaming/blob/main/docs/source/getting_started/faqs_and_tips.md

Example of how to instantiate a StreamingDataset and retrieve the number of unique samples it contains. This count excludes any upsampling or downsampling.

```python
# Instantiate a StreamingDataset however you would like
dataset = StreamingDataset(
    ...
)
# Retrieves the number of unique samples -- no up or down sampling applied
num_dataset_samples = dataset.size
```

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

### Configure AWS CLI Credentials

Source: https://github.com/mosaicml/streaming/blob/main/docs/source/how_to_guides/configure_cloud_storage_credentials.md

Install the AWS CLI and configure your AWS credentials using the `aws configure` command. This creates the necessary config and credential files.

```bash
python -m pip install awscli
aws configure
```

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

### Prepare Data with MDSWriter

Source: https://github.com/mosaicml/streaming/blob/main/README.md

Example of preparing data using MDSWriter. This involves specifying the output directory, column types, and compression format. Ensure the `data_dir` points to your desired local or remote storage location.

```python
import numpy as np
from PIL import Image
from streaming import MDSWriter

# Local or remote directory in which to store the compressed output files
data_dir = 'path-to-dataset'

# A dictionary mapping input fields to their data types
columns = {
    'image': 'jpeg',
    'class': 'int'
}

# Shard compression, if any
compression = 'zstd'

```

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

### Install img2dataset for Data Crawling

Source: https://github.com/mosaicml/streaming/blob/main/streaming/multimodal/convert/laion/laion400m/README.md

Installs the img2dataset package, version 1.41.0, which is used for crawling and downloading image-text pairs. Optional performance monitoring tools are also listed.

```bash
# Used for crawling.
pip3 install img2dataset==1.41.0

# Optional performance monitoring.
apt install bwm-ng htop iotop
```

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

### Create DataLoader for StreamingDataset

Source: https://github.com/mosaicml/streaming/blob/main/README.md

Illustrates how to create a PyTorch DataLoader for a StreamingDataset. This setup allows for iterating over the dataset with specified worker configurations.

```python
dataset = StreamingDataset(...)
dl = DataLoader(dataset, num_workers=...)
```

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

### Initialize StreamingDataset with Multiple Streams

Source: https://github.com/mosaicml/streaming/blob/main/README.md

Configure a StreamingDataset by providing a list of streams, each with a remote path and a proportion. This setup is used for distributing data from various sources like S3 and GCS.

```python
streams = [
  Stream(remote='s3://datasets/c4', proportion=0.4),
  Stream(remote='s3://datasets/github', proportion=0.1),
  Stream(remote='gcs://datasets/my_internal', proportion=0.5),
]

dataset = StreamingDataset(
  streams=streams,
  samples_per_epoch=1e8,
)
```

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

### Initialize global variables for dataset conversion

Source: https://github.com/mosaicml/streaming/blob/main/docs/source/preparing_datasets/parallel_dataset_conversion.ipynb

Define the output root directory, the dataset to be processed (represented by a list of numbers in this example), and the column names and types for the MDS dataset.

```python
out_root = './data'
# This could be a list of URLs needs to download
dataset = [i for i in range(25)]
columns = {'number': 'int'}
```

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

### Initialize StreamingDataset with Single Stream Configuration

Source: https://github.com/mosaicml/streaming/blob/main/docs/source/getting_started/main_concepts.md

Configure a StreamingDataset for a single data source by directly specifying the remote and local storage paths. This is a simpler setup for datasets residing in a single location.

```python
dataset = StreamingDataset(remote='s3://some/path', local='/local/path', batch_size=4)
```

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

### Build and View Docs Locally

Source: https://github.com/mosaicml/streaming/blob/main/STYLE_GUIDE.md

Commands to activate the virtual environment, navigate to the docs directory, clean previous builds, build HTML documentation, and run a local server to view the docs.

```bash
source path/to/streaming_venv/bin/activate  # activate your streaming virtual env
cd streaming/docs  # cd to the docs folder inside your streaming clone
make clean  # Cleans the artifacts and remove source/api_reference folder
make html   # build the docs
make host   # Run the docs locally
```

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

### Build and View Documentation Locally

Source: https://github.com/mosaicml/streaming/blob/main/CONTRIBUTING.md

Compiles and serves the project documentation locally. This is mandatory if you have made documentation changes.

```bash
pip install -e '.[docs]'
cd docs
make clean && make html
make host   # open the output link in a browser.
```

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

### Run Doctests Locally

Source: https://github.com/mosaicml/streaming/blob/main/STYLE_GUIDE.md

Commands to activate the virtual environment, navigate to the docs directory, clean previous builds, build HTML documentation (required for doctest identification), and then run the doctests.

```bash
source path/to/streaming_venv/bin/activate  # activate your streaming virtual env
cd streaming/docs  # cd to the docs folder inside your streaming clone
make clean  # Cleans the artifacts and remove source/api_reference folder
make html  # the html build must be completed first to ensure all doctests are identified
make doctest 2>/dev/null # For more verbosity, do not direct stderr to /dev/null
```

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

### Initialize Model, Loss, and Optimizer

Source: https://github.com/mosaicml/streaming/blob/main/examples/cifar10.ipynb

Initializes the CNN model, moves it to the appropriate device, and sets up the CrossEntropyLoss criterion and SGD optimizer with specified parameters.

```python
model = Net()
model = model.to(device)
criterion = nn.CrossEntropyLoss()
optimizer = torch.optim.SGD(model.parameters(), lr=0.001, momentum=0.9, weight_decay=5e-4)
```

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

### Configure Pre-downloaded Shards

Source: https://github.com/mosaicml/streaming/blob/main/docs/source/dataset_configuration/shard_retrieval.md

Set `predownload` to pre-fetch shards for upcoming samples, improving performance by reducing I/O during training. The default value is usually sufficient.

```python
dataset = StreamingDataset(
    ...
    predownload = 8,    # each worker will download shards for up to 8 samples ahead
    ...
)
```

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

### Train DeepLabV3 with Composer

Source: https://github.com/mosaicml/streaming/blob/main/examples/facesynthetics.ipynb

Set up a DeepLabV3 model and optimizer, then configure and run the Composer trainer with streaming dataloaders. Ensure the trainer is initialized with the model, dataloaders, training duration, optimizer, and device.

```python
model = composer_deeplabv3(
    num_classes=num_classes, 
    backbone_arch='resnet101', 
    backbone_weights='IMAGENET1K_V2',
    sync_bn=False)
optimizer = DecoupledAdamW(model.parameters(), lr=1e-3)

trainer = Trainer(
    model=model,
    train_dataloader=train_dataloader,
    eval_dataloader=test_dataloader,
    max_duration=train_epochs,
    optimizers=optimizer,
    device=device
)

start_time = time.perf_counter()
trainer.fit()
end_time = time.perf_counter()
print(f"It took {end_time - start_time:0.4f} seconds to train")
```

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

### Data Validation Example

Source: https://github.com/mosaicml/streaming/blob/main/STYLE_GUIDE.md

Use this pattern for data validation instead of assert, as asserts can be disabled.

```python
if parameter is None:
    raise ValueError("parameter must be specified and cannot be None")
```

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

### Uninstall mosaicml-streaming

Source: https://github.com/mosaicml/streaming/blob/main/examples/spark_dataframe_to_MDS.ipynb

Uninstalls the mosaicml-streaming package. This might be used to ensure a clean installation or to revert to a previous version.

```python
%pip uninstall -y mosaicml-streaming
```

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

### Initialize StreamingInsideWebVid Dataset

Source: https://github.com/mosaicml/streaming/blob/main/README.md

Import and initialize the StreamingInsideWebVid dataset class for training. Specify local cache and remote storage paths, and enable shuffling.

```python
from streaming.multimodal import StreamingInsideWebVid

local = '/tmp/path-to-dataset'
remote = 's3://my-bucket/path-to-dataset'

dataset = StreamingInsideWebVid(local=local, remote=remote, shuffle=True)
```

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

### Load MDS Dataset with StreamingDataset

Source: https://github.com/mosaicml/streaming/blob/main/docs/source/preparing_datasets/spark_dataframe_to_mds.ipynb

Load the MDS dataset using `StreamingDataset` and create a DataLoader. This example demonstrates how to initialize the dataset and iterate through batches.

```python
from torch.utils.data import DataLoader
import streaming
from streaming import StreamingDataset

streaming.base.util.clean_stale_shared_memory()

dataset = StreamingDataset(local=out_path, remote=None, batch_size=2, predownload=4)

dataloader = DataLoader(dataset, batch_size=2, num_workers=1)

for i, data in enumerate(dataloader):
    print(data)
    if i == 10:
        break
```

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

### Configure Global Settings for Dataset and Training

Source: https://github.com/mosaicml/streaming/blob/main/examples/facesynthetics.ipynb

Sets up global variables for dataset paths (input, output, local), shuffling preferences, number of classes, shard size limit, progress bar usage, training ratio, batch size, and device selection.

```python
# the location of our dataset
in_root = "./dataset"

# the location of the "remote" streaming dataset (`sds`). 
# Upload `out_root` to your cloud storage provider of choice.
out_root = "./sds"
out_train = "./sds/train"
out_test = "./sds/test"

# the location to download the streaming dataset during training
local = './local'
local_train = './local/train'
local_test = './local/test'

# toggle shuffling in dataloader
shuffle_train = True
shuffle_test = False

# possible values for a pixel in the annotation image to take
num_classes = 20

# shard size limit, in bytes
size_limit = 1 << 25

# show a progress bar while downloading
use_tqdm = True

# ratio of training data to test data
training_ratio = 0.9

# training batch size
batch_size = 2 # this is the smallest batch size possible, 
               # increase this if your machine can handle it.

# training hardware parameters
device = "gpu" if torch.cuda.is_available() else "cpu"

```

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

### Get StreamingDataset Statistics

Source: https://github.com/mosaicml/streaming/blob/main/docs/source/how_to_guides/synthetic_nlp.ipynb

Retrieve utility information about the dataset, such as the total number of samples, the number of shard files, and the distribution of samples per shard.

```python
# Get the total number of samples
print(f'Total number of samples: {train_dataset.num_samples}')

# Get the number of shard files
print(f'Total number of shards: {len(train_dataset.shards)}')

# Get the number of samples inside each shard files.
# Number of samples in each shard can vary based on each sample size.
print(f'Number of samples inside each shards: {train_dataset.samples_per_shard}')

```

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

### Run Unit and Documentation Tests

Source: https://github.com/mosaicml/streaming/blob/main/CONTRIBUTING.md

Executes all unit tests and documentation tests locally. The `ulimit` command is a workaround for 'Too many open files' issues.

```bash
ulimit -n unlimited # Workaround: To overcome 'Too many open files' issues since streaming uses atexit handler to close file descriptor at the end.

pytest -vv -s . # run all the unittests
cd docs && make clean && make doctest # run doctests
```

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

### Write and Inspect MDS Dataset

Source: https://github.com/mosaicml/streaming/blob/main/docs/source/preparing_datasets/basic_dataset_conversion.md

Use MDSWriter to create a dataset with fixed array shape and datatype, then inspect it with StreamingDataset. Ensure you have numpy and mosaicml installed.

```python
from streaming.base.writer import MDSWriter
from streaming.base.dataset import StreamingDataset
import numpy as np

with MDSWriter(out='my_dataset3/',
               columns={'my_array': 'ndarray:int16:3,3,3'}) as out:
    for i in range(42):
        # Shape is fixed
        shape = 3, 3, 3
        # Datatype is fixed
        my_array = np.random.normal(0, 100, shape).astype(np.int16)
        out.write({'my_array': my_array})

# Inspect dataset
dataset = StreamingDataset(local='my_dataset3/', batch_size=1)
for i in range(dataset.num_samples):
    print(dataset[i])
```

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

### Specify Hash Algorithms for MDS Shard Verification

Source: https://github.com/mosaicml/streaming/blob/main/docs/source/preparing_datasets/basic_dataset_conversion.md

Optionally, provide a list of hash algorithms to be used for verifying the integrity of each shard. SHA1 is specified in this example.

```python
hashes = ['sha1'] # Use only SHA1 hashing on each shard
```

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

### Initialize Global Variables for MDS Conversion

Source: https://github.com/mosaicml/streaming/blob/main/docs/source/preparing_datasets/parallel_dataset_conversion.ipynb

Set up root output directory, number of groups, and number of processes for parallel conversion.

```python
out_root = './group_data'
num_groups = 4
num_process = 2
```

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

### MDS Output Directory Structure

Source: https://github.com/mosaicml/streaming/blob/main/docs/source/preparing_datasets/basic_dataset_conversion.md

An example of the file structure created in the output directory after writing a dataset to MDS format. It includes an index file and shard files.

```bash
dirname
├── index.json
├── shard.00000.mds.zstd
└── shard.00001.mds.zstd
```

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

### Write Dataset to MDS Format using MDSWriter

Source: https://github.com/mosaicml/streaming/blob/main/docs/source/preparing_datasets/basic_dataset_conversion.md

Instantiate MDSWriter with the specified parameters and write samples from a dataset by iterating over it. This example writes a synthetic classification dataset.

```python
from streaming.base import MDSWriter

dataset = RandomClassificationDataset()
with MDSWriter(out=output_dir, columns=columns, compression=compression, hashes=hashes, size_limit=limit) as out:
    for x, y in dataset:
        out.write({'x': x, 'y': y})
```

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

### Define a Synthetic Classification Dataset

Source: https://github.com/mosaicml/streaming/blob/main/docs/source/preparing_datasets/basic_dataset_conversion.md

Create a custom dataset class that generates synthetic data for classification tasks. This class is used as an example for writing to MDS format.

```python
import numpy as np

class RandomClassificationDataset:
    """Classification dataset drawn from a normal distribution.

    Args:
        shape: data sample dimensions (default: (10,))
        size: number of samples (default: 10000)
        num_classes: number of classes (default: 2)
    """

    def __init__(self, shape=(10,), size=10000, num_classes=2):
        self.size = size
        self.x = np.random.randn(size, *shape)
        self.y = np.random.randint(0, num_classes, size)

    def __len__(self):
        return self.size

    def __getitem__(self, index: int):
        return self.x[index], self.y[index]
```

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

### Initialize StreamingDataset with DataLoader

Source: https://github.com/mosaicml/streaming/blob/main/docs/source/index.md

Demonstrates how to initialize a StreamingDataset and use it with PyTorch's DataLoader. Ensure the remote path is correctly specified.

```python
from torch.utils.data import DataLoader
from streaming import StreamingDataset

dataloader = DataLoader(dataset=StreamingDataset(remote='s3://...', batch_size=1))
```

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

### Get Train and Validation Dataset Splits

Source: https://github.com/mosaicml/streaming/blob/main/docs/source/how_to_guides/synthetic_nlp.ipynb

Generates the complete training and validation datasets by combining number generation and sample creation. Returns two lists of dictionaries.

```python
def get_dataset(num_train: int, num_val: int) -> Tuple[List[Dict[str, Any]], List[Dict[str, Any]]]:
    """Generate a number-saying dataset of the given size.

    Args:
        num_train (int): Number of training samples.
        num_val (int): Number of validation samples.

    Returns:
        Tuple[List[Dict[str, Any]], List[Dict[str, Any]]]: The two generated splits.
    """
    train_nums, val_nums = get_numbers(num_train, num_val)
    train_samples = generate_samples(train_nums)
    val_samples = generate_samples(val_nums)
    return train_samples, val_samples
```

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

### Import necessary libraries

Source: https://github.com/mosaicml/streaming/blob/main/examples/synthetic_nlp.ipynb

Import core libraries for data manipulation, PyTorch, and streaming dataset utilities.

```python
import os
import shutil
from typing import Any, Dict, List, Tuple

import numpy as np
import torch
from torch.utils.data import DataLoader
from tqdm import tqdm
```

```python
from streaming import MDSWriter, StreamingDataset
```

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

### Wrap StreamingDataset in PyTorch DataLoader

Source: https://github.com/mosaicml/streaming/blob/main/examples/synthetic_nlp.ipynb

Wrap a StreamingDataset instance in a standard PyTorch DataLoader for efficient batching and parallel data loading during model training. Ensure PyTorch is installed.

```python
train_dataloader = DataLoader(train_dataset, batch_size=batch_size)
val_dataloader = DataLoader(val_dataset, batch_size=batch_size)
```

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

### Load Dataset from Local Path

Source: https://github.com/mosaicml/streaming/blob/main/docs/source/dataset_configuration/shard_retrieval.md

Use this when your dataset shards are already available on local disk accessible by your GPUs.

```python
dataset = StreamingDataset(
    local = '/local/dataset',    # dataset shards are already locally available at this path
)
```

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

### Create Training Data Split

Source: https://github.com/mosaicml/streaming/blob/main/streaming/text/convert/enwiki/mds/README.md

Run the Python script to create the training data split. This script calls another Python script for data creation.

```bash
python3 make_train_parallel.py
```

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

### Run Pre-commit Hooks

Source: https://github.com/mosaicml/streaming/blob/main/CONTRIBUTING.md

Runs pre-commit hooks to lint and format staged files. Use the --all-files flag to run on all files in the repository.

```bash
git add <file1> <file2>
pre-commit run
```

```bash
pre-commit run --all-files
```

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

### Import Necessary Libraries

Source: https://github.com/mosaicml/streaming/blob/main/docs/source/preparing_datasets/spark_dataframe_to_mds.ipynb

Import core libraries for Spark, data manipulation, and streaming dataset conversion.

```python
import os
import shutil
from typing import Any, Sequence, Dict, Iterable, Optional
from pyspark.sql import SparkSession
import pandas as pd
import numpy as np
from tempfile import mkdtemp
import datasets as hf_datasets
from transformers import AutoTokenizer, PreTrainedTokenizerBase
```

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

### Upload Dataset to S3 Bucket using AWS CLI

Source: https://github.com/mosaicml/streaming/blob/main/README.md

Upload a local dataset directory recursively to an S3 bucket using the AWS CLI. Ensure you have the AWS CLI installed and configured.

```bash
aws s3 cp --recursive path-to-dataset s3://my-bucket/path-to-dataset
```

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

### Prepare MDS Conversion Parameters

Source: https://github.com/mosaicml/streaming/blob/main/docs/source/preparing_datasets/spark_dataframe_to_mds.ipynb

Set up the output path and column mapping for MDS conversion. Ensure the `columns` field correctly maps the output from your iterable function.

```python
out_path = os.path.join(local_dir, 'mds')
shutil.rmtree(out_path, ignore_errors=True)

mds_kwargs = {'out': out_path, 'columns': {'tokens': 'bytes'}}

udf_kwargs = {
    'concat_tokens': 4,
    'tokenizer': 'EleutherAI\/gpt-neox-20b',
    'eos_text': '<|endoftext|>',
    'compression': 'zstd',
    'split': 'train',
    'no_wrap': False,
    'bos_text': '',
}
```

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

### Set Shard Size Limit for MDSWriter

Source: https://github.com/mosaicml/streaming/blob/main/docs/source/preparing_datasets/basic_dataset_conversion.md

Optionally, set a maximum size for each shard. A new shard will be created once this limit is reached. '10kb' is used for this small example; larger values are recommended for production.

```python
# Here we use a human-readable string, but we could also
# pass in an int specifying the number of bytes.
limit = '10kb'
```

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

### Build and Use a StreamingDataset with PyTorch DataLoader

Source: https://github.com/mosaicml/streaming/blob/main/README.md

Create a streaming dataset from remote and local paths, shuffle it, and access individual samples. Then, create a PyTorch DataLoader for batching.

```python
from torch.utils.data import DataLoader
from streaming import StreamingDataset

# Remote path where full dataset is persistently stored
remote = 's3://my-bucket/path-to-dataset'

# Local working dir where dataset is cached during operation
local = '/tmp/path-to-dataset'

# Create streaming dataset
dataset = StreamingDataset(local=local, remote=remote, shuffle=True)

# Let's see what is in sample #1337...
sample = dataset[1337]
img = sample['image']
cls = sample['class']

# Create PyTorch DataLoader
dataloader = DataLoader(dataset)
```

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

### Get StreamingDataset Utility Information

Source: https://github.com/mosaicml/streaming/blob/main/examples/synthetic_nlp.ipynb

Retrieve utility information about a StreamingDataset, such as the total number of samples, the number of shard files, and the number of samples per shard. This is helpful for understanding dataset size and distribution.

```python
# Get the total number of samples
print(f'Total number of samples: {train_dataset.num_samples}')

# Get the number of shard files
print(f'Total number of shards: {len(train_dataset.shards)}')

# Get the number of samples inside each shard files.
# Number of samples in each shard can vary based on each sample size.
print(f'Number of samples inside each shards: {train_dataset.samples_per_shard}')
```

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

### Download and Convert LAION-400M Data to MDS

Source: https://github.com/mosaicml/streaming/blob/main/streaming/multimodal/convert/laion/laion400m/README.md

Runs a script to download data from the web, saving it into parquet files, and then converts it to Mosaic Dataset Shard (MDS) format. It is recommended to run this concurrently with the conversion and upload script to manage disk usage.

```bash
./streaming/multimodal/convert/laion/laion400m/download_data.sh
```

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

### Example index.json Structure for MDS Shards

Source: https://github.com/mosaicml/streaming/blob/main/docs/source/getting_started/main_concepts.md

This JSON structure shows the metadata for multiple MDS shards, including column information, sample counts, and shard sizes. It is used by StreamingDataset to understand the composition of a data stream.

```json
{
    "shards": [
        {
            "column_encodings": ["bytes"],
            "column_names": ["tokens"],
            "column_sizes": [null],
            "compression": null,
            "format": "mds",
            "hashes": [],
            "raw_data": {
                "basename": "shard.00000.mds",
                "bytes": 67092637,
                "hashes": {}
            },
            "samples": 4093,
            "size_limit": 67108864,
            "version": 2,
            "zip_data": null
        },
        {
            "column_encodings": ["bytes"],
            "column_names": ["tokens"],
            "column_sizes": [null],
            "compression": null,
            "format": "mds",
            "hashes": [],
            "raw_data": {
                "basename": "shard.00001.mds",
                "bytes": 67092637,
                "hashes": {}
            },
            "samples": 4093,
            "size_limit": 67108864,
            "version": 2,
            "zip_data": null
        }
    ]
}
```

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

### Train with Streaming Dataloaders

Source: https://github.com/mosaicml/streaming/blob/main/docs/source/how_to_guides/cifar10.ipynb

Initiates the training process by iterating through epochs, calling the `fit` and `eval` functions, and printing the training and validation metrics.

```python
for epoch in range(train_epochs):
    train_epoch_loss, train_epoch_accuracy = fit(model, train_dataloader)
    print(f'epoch: {epoch+1}/{train_epochs} Train Loss: {train_epoch_loss:.4f}, Train Acc: {train_epoch_accuracy:.2f}')
    val_epoch_loss, val_epoch_accuracy = eval(model, test_dataloader)
    print(f'epoch: {epoch+1}/{train_epochs} Val Loss: {val_epoch_loss:.4f}, Val Acc: {val_epoch_accuracy:.2f}')
```

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

### Configure Sample Replication for Distributed Training

Source: https://github.com/mosaicml/streaming/blob/main/docs/source/dataset_configuration/replication_and_sampling.md

Set the `replication` parameter to control how many consecutive devices see the same samples in each batch. Useful for TP/SP.

```python
dataset = StreamingDataset(
    ...
    replication = 4,    # Every 4 GPUs will see the same samples.
    ...
)
```

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

### List Directory Structure After Conversion

Source: https://github.com/mosaicml/streaming/blob/main/docs/source/preparing_datasets/parallel_dataset_conversion.ipynb

Use the `%ll` magic command to display the detailed directory structure, showing the sub-directories created for MDS shards, each containing an `index.json` file and shard files.

```python
%ll $out_root
```

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

### Set Alipan Encryption Credentials (Shell)

Source: https://github.com/mosaicml/streaming/blob/main/docs/source/how_to_guides/configure_cloud_storage_credentials.md

Configure Alipan encryption password and type in the run environment using shell for encrypting data during uploads.

```bash
export ALIPAN_ENCRYPT_PASSWORD='encryption_key'

# For uploading, the encryption type must be set as one of `Simple`, `ChaCha20`, `AES256CBC`
# When downloading, this environment variable is not required
export ALIPAN_ENCRYPT_TYPE='AES256CBC'
```

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

### Create Synthetic Dataset

Source: https://github.com/mosaicml/streaming/blob/main/docs/source/how_to_guides/synthetic_nlp.ipynb

Initializes and generates the synthetic dataset with specified training and validation sample counts. Prints the generation progress.

```python
# Number of training and validation samples
num_train_samples = 10_000 # 10k samples
num_val_samples = 2000    # 2k samples

# Create the samples.
print(f'Generating synthetic dataset ({num_train_samples} train, {num_val_samples} val)...')
train_samples, val_samples = get_dataset(num_train_samples, num_val_samples)
splits = [
    ('train', train_samples),
    ('val', val_samples)
]
```

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

### Cleanup Local Files

Source: https://github.com/mosaicml/streaming/blob/main/docs/source/how_to_guides/cifar10.ipynb

Removes temporary directories created during the tutorial to free up disk space. Uses `shutil.rmtree` for recursive directory deletion.

```python
shutil.rmtree(out_root, ignore_errors=True)
shutil.rmtree(in_root, ignore_errors=True)
shutil.rmtree(local, ignore_errors=True)
```

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

### Configure global settings for streaming dataset

Source: https://github.com/mosaicml/streaming/blob/main/examples/synthetic_nlp.ipynb

Define local and remote paths for dataset storage, and configure shuffling and batch size for the dataloader.

```python
# the location of the "remote" streaming dataset (`sds`). 
# Upload `out_root` to your cloud storage provider of choice. If `out_root` is a cloud provider
# path, shard files are automatically uploaded.
out_root = "./sds"
out_train = "./sds/train"
out_val = "./sds/val"

# the location to download the streaming dataset during training
local = './local'
local_train = './local/train'
local_val = './local/val'

# toggle shuffling in dataloader
shuffle_train = True
shuffle_val = False

# training batch size
batch_size = 512
```

```python
# upload location for the dataset splits (change this if you want to upload to a different location, for example, AWS S3 bucket location)
upload_location = None

if upload_location is None:
    upload_train_location = None
    upload_val_location = None
else:
    upload_train_location = os.path.join(upload_location, 'train')
    upload_val_location = os.path.join(upload_location, 'val')
```

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

### Save Samples as Shards with MDSWriter

Source: https://github.com/mosaicml/streaming/blob/main/README.md

Use MDSWriter to save samples into sharded files. Ensure columns and compression are configured appropriately.

```python
from streaming import MDSWriter

with MDSWriter(out=data_dir, columns=columns, compression=compression) as out:
    for i in range(10000):
        sample = {
            'image': Image.fromarray(np.random.randint(0, 256, (32, 32, 3), np.uint8)),
            'class': np.random.randint(10),
        }
        out.write(sample)
```

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

### Set Huggingface Hub Token (Python)

Source: https://github.com/mosaicml/streaming/blob/main/docs/source/how_to_guides/configure_cloud_storage_credentials.md

Configure the Huggingface Hub token in the run environment using Python for authentication.

```python
import os
os.environ['HF_TOKEN'] = 'EXAMPLEFODNN7EXAMPLE'
```

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

### Import necessary libraries

Source: https://github.com/mosaicml/streaming/blob/main/docs/source/how_to_guides/synthetic_nlp.ipynb

Import core Python libraries and PyTorch components for data handling and model training.

```python
import os
import shutil
from typing import Any, Dict, List, Tuple

import numpy as np
import torch
from torch.utils.data import DataLoader
from tqdm import tqdm
```

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

### Inspect First Train and Validation Samples

Source: https://github.com/mosaicml/streaming/blob/main/docs/source/how_to_guides/synthetic_nlp.ipynb

Prints the first sample from both the training and validation datasets to allow for inspection of the generated data structure.

```python
print(f'Train sample: {train_samples[0]}')
print(f'Val sample: {val_samples[0]}')
```

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

### Import Composer Trainer and Optimizer

Source: https://github.com/mosaicml/streaming/blob/main/examples/facesynthetics.ipynb

Imports the `Trainer` class and the `DecoupledAdamW` optimizer from the `composer` library, along with the `composer_deeplabv3` model.

```python
from composer import Trainer
from composer.models import composer_deeplabv3
from composer.optim import DecoupledAdamW
```

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

### Set Huggingface Hub Token (Shell)

Source: https://github.com/mosaicml/streaming/blob/main/docs/source/how_to_guides/configure_cloud_storage_credentials.md

Configure the Huggingface Hub token in the run environment using shell for authentication.

```bash
export HF_TOKEN='EXAMPLEFODNN7EXAMPLE'
```

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

### Set Alipan Refresh Token (Shell)

Source: https://github.com/mosaicml/streaming/blob/main/docs/source/how_to_guides/configure_cloud_storage_credentials.md

Configure the Alipan refresh token in the run environment using shell for authentication.

```bash
export ALIPAN_WEB_REFRESH_TOKEN='refresh_token'
```

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

### Instantiate Streaming Datasets and DataLoaders

Source: https://github.com/mosaicml/streaming/blob/main/docs/source/how_to_guides/cifar10.ipynb

Creates instances of the custom `CIFAR10Dataset` for training and testing, then wraps them in PyTorch `DataLoader` objects for batching and iteration during training.

```python
train_dataset = CIFAR10Dataset(remote_train, local_train, shuffle_train, batch_size=batch_size, transforms=transformation)
test_dataset  = CIFAR10Dataset(remote_test, local_test, shuffle_test, batch_size=batch_size, transforms=transformation)

train_dataloader = DataLoader(train_dataset, batch_size=batch_size)
test_dataloader = DataLoader(test_dataset, batch_size=batch_size)
```

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

### Clean Up Generated Files

Source: https://github.com/mosaicml/streaming/blob/main/docs/source/how_to_guides/synthetic_nlp.ipynb

Remove the local and remote directories created during the tutorial to free up disk space.

```python
shutil.rmtree(out_root, ignore_errors=True)
shutil.rmtree(local, ignore_errors=True)

```

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

### Access a Specific Sample by Index

Source: https://github.com/mosaicml/streaming/blob/main/README.md

Demonstrates random access to dataset samples. If a sample is not yet downloaded, accessing it via `dataset[i]` will initiate the download.

```python
dataset = StreamingDataset(...)
sample = dataset[19543]
```

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

### Set Alipan Encryption Credentials (Python)

Source: https://github.com/mosaicml/streaming/blob/main/docs/source/how_to_guides/configure_cloud_storage_credentials.md

Configure Alipan encryption password and type in the run environment using Python for encrypting data during uploads.

```python
import os
os.environ['ALIPAN_ENCRYPT_PASSWORD'] = 'encryption_key'

# For uploading, the encryption type must be set as one of `Simple`, `ChaCha20`, `AES256CBC`
# When downloading, this environment variable is not required
os.environ['ALIPAN_ENCRYPT_TYPE'] = 'AES256CBC'
```

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

### Instantiate Streaming Datasets and DataLoaders

Source: https://github.com/mosaicml/streaming/blob/main/examples/facesynthetics.ipynb

Creates instances of the custom `FaceSynthetics` dataset for training and testing, then wraps them in PyTorch `DataLoader` objects for batching and iteration.

```python
dataset_train = FaceSynthetics(remote_train, local_train, shuffle_train, batch_size=batch_size)
dataset_test  = FaceSynthetics(remote_test, local_test, shuffle_test, batch_size=batch_size)

train_dataloader = DataLoader(dataset_train, batch_size=batch_size)
test_dataloader = DataLoader(dataset_test, batch_size=batch_size)
```

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

### Upload Dataset to S3 (Optional)

Source: https://github.com/mosaicml/streaming/blob/main/docs/source/how_to_guides/cifar10.ipynb

This is a commented-out command to upload the entire streaming dataset directory to an AWS S3 bucket. Uncomment and modify `upload_location` if you wish to use this feature.

```bash
# !aws s3 cp $out_root $upload_location --recursive
```

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

### Specify Absolute Samples with `choose`

Source: https://github.com/mosaicml/streaming/blob/main/docs/source/dataset_configuration/mixing_data_sources.md

Use the `choose` argument to specify the exact number of samples to take from a stream per epoch when mixing datasets.

```python
stream_A = Stream(
    remote = 's3://stream_A_remote',
    local = '/tmp/stream_A',
    choose = 250,
)
```

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

### Define global settings for streaming dataset

Source: https://github.com/mosaicml/streaming/blob/main/docs/source/how_to_guides/synthetic_nlp.ipynb

Configure paths for local and remote storage, shuffling options, and batch size for the streaming dataset.

```python
# the location of the "remote" streaming dataset (`sds`). 
# Upload `out_root` to your cloud storage provider of choice. If `out_root` is a cloud provider
# path, shard files are automatically uploaded.
out_root = "./sds"
out_train = "./sds/train"
out_val = "./sds/val"

# the location to download the streaming dataset during training
local = './local'
local_train = './local/train'
local_val = './local/val'

# toggle shuffling in dataloader
shuffle_train = True
shuffle_val = False

# training batch size
batch_size = 512
```

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

### Cleanup Generated Files

Source: https://github.com/mosaicml/streaming/blob/main/examples/facesynthetics.ipynb

Remove temporary files and directories created during the tutorial to free up disk space. This includes output roots, input roots, local directories, and the dataset archive.

```python
shutil.rmtree(out_root, ignore_errors=True)
shutil.rmtree(in_root, ignore_errors=True)
shutil.rmtree(local, ignore_errors=True)
if os.path.exists(dataset_archive):
    os.remove(dataset_archive)
```

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

### Set Alipan Refresh Token (Python)

Source: https://github.com/mosaicml/streaming/blob/main/docs/source/how_to_guides/configure_cloud_storage_credentials.md

Configure the Alipan refresh token in the run environment using Python for authentication.

```python
import os
os.environ['ALIPAN_WEB_REFRESH_TOKEN'] = 'refresh_token'
```

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

### Load WebVid Dataset with StreamingInsideWebVid

Source: https://github.com/mosaicml/streaming/blob/main/docs/source/getting_started/quick_start.md

Import and instantiate the StreamingInsideWebVid dataset class to load the WebVid dataset for training. Specify local and remote directories, batch size, and shuffle option.

```python
from streaming.multimodal import StreamingInsideWebVid
dataset = StreamingInsideWebVid(local=local, remote=remote, batch_size=1, shuffle=True)

```