Skip to main content

AI Dataschema User Manual - Basic Usage

This guide walks you through the basic features and functionalities of AI Dataschema. The package is available in CAIP notebook environments and can be used to create standardized datasets, load and query data from datasets, and prepare data for machine learning workflows.

What is the main idea behind AI Dataschema?​

AI Dataschema is BMW's internal, open-source-based data standard designed to streamline AI development. It unifies support for multiple data types such as tables, images, point clouds, and text within a single standardized schema.

Through a user-friendly data API, standardized datasets can be created for internal projects or organization-wide use. The schema supports data conversion, synchronization of diverse data sources, and the training and deployment of machine learning models. This guide focuses on the primary capabilities: saving data, loading data, synchronizing tables, and preparing data for machine learning. It is compatible with both Linux and Windows.

Create an AI Dataset​

Working with AI Dataschema begins with creating the base directory, referred to as a dataset. To create a dataset, invoke the add_dataset() method of the AIdataset class. This method accepts a required name and an optional metadata argument. You may also specify a path for saving the dataset. If no path is provided, the dataset is saved in the user's home directory.

The metadata parameter is optional. You can attach dataset information using either a Python dictionary or a JSON or YAML file.

from pathlib import Path
from AIdataschema import AIdataset

path = Path.home()

dataset = AIdataset.add_dataset(
    name="AIdata_manual",
    path=path,
    metadata={
        "description": "This dataset contains data for demo purposes.",
        "owner": "Max Mustermann",
    },
    # Or by uploading a JSON or YAML file:
    # metadata="path/to/your_metadata.yaml",
)

Create Multiple Streams and Upload Data​

Data is organized into streams within a dataset. Adding data to an existing dataset is performed with add_stream(), where you specify the stream name, the path to the local directory or data file, the converter type, and optional metadata. If desired, you may set version_name to assign a specific name to the uploaded data. Each write automatically increments the dataset version, and you can revert to a previous version at any time.

When reading in batches later, it is important that data is written in the intended order if grouping depends on contiguous records. If related records are spread across multiple batches, grouped processing may yield partial results per batch instead of a single combined result.

The following example illustrates how to convert data into AI Dataschema using Parquet input.

from pathlib import Path
from AIdataschema import AIdataset
from AIdataschema import ParquetConverter

path = Path.home()
path_to_your_dataset = path / "AIdata_manual"

dataset = AIdataset.load_dataset(path=path_to_your_dataset)

STREAM_NAME_1 = "rsu_data"
RSU_METADATA = "path/to/dataset_metadata.json"
RSU_VERSION_MESSAGE = "Initial stream upload of RSU data"
PARQUET_FILES = "path/to/first_stream"

dataset.add_stream(
    stream_name=STREAM_NAME_1,
    path=PARQUET_FILES,
    converter_type=ParquetConverter,
    metadata=RSU_METADATA,
    version_name=RSU_VERSION_MESSAGE,
)

You can also add a second stream with a custom converter.

from pathlib import Path
from AIdataschema import AIdataset
from ext_converters.csv_converter import MyCSVConverter

path = Path.home()
path_to_your_dataset = path / "AIdata_manual"
dataset = AIdataset.load_dataset(path=path_to_your_dataset)

STREAM_NAME_2 = "model_mapping"
MAPPING_METADATA = {
    "description": "Mapping: internal model designation",
    "owner": "Source: https://www.bimmerarchiv.de/e-code/",
}
MAPPING_VERSION_MESSAGE = "Initial stream upload of model mapping"
CSV_FILE = "path/to/second_stream/Modellbezeichnungen.csv"

dataset.add_stream(
    stream_name=STREAM_NAME_2,
    path=CSV_FILE,
    converter_type=MyCSVConverter,
    metadata=MAPPING_METADATA,
    version_name=MAPPING_VERSION_MESSAGE,
)

Metainformation​

Use get_metadata() to inspect the dataset's metadata. This shows which streams are included and the dataset's history, including version notes.

print(dataset.get_metadata())

If you add metadata directly into stream columns, retrieve it together with the rest of the stream columns when reading the data.

Retrieve Stream Data via the API​

To read data from a stream, first load the dataset with load_dataset() and then load the stream with load_stream(), optionally specifying a version. If you omit the version, all uploaded data for the stream are returned. Use exact=True to retrieve only the exact version, or exact=False to include older versions up to that point.

Once the stream is loaded, invoke read_data_stream(). You may specify columns, an output format, and a filter expression using PyArrow syntax. The returned object is a DataObject. Its main attribute, .df, contains the data in the requested format. You can also call get_metadata() on the DataObject to retrieve metadata for the returned data.

import pyarrow.compute as pc
from AIdataschema import AIdataset

dataset = AIdataset.load_dataset(path=path_to_your_dataset)
stream = dataset.load_stream(stream_name="rsu_data", version=1, exact=True)

selected_columns = [
    "uuid",
    "dtc_vehicle_timestamp",
    "dtc_360_id",
    "model_range",
]

filter_expr = pc.field("model_range") == "G05"

result = stream.read_data_stream(
    columns=selected_columns,
    filter_expr=filter_expr,
    output="pandas",
)

result.df
result.get_metadata(50)["description"]

Retrieve Large Stream Data via the API​

To load data incrementally, use read_data_batch_stream() to obtain a generator when a stream cannot be fully loaded into memory. Specify batch_size for the number of rows per iteration. The remaining parameters are the same as for read_data_stream().

from AIdataschema import AIdataset
import pyarrow.compute as pc

dataset = AIdataset.load_dataset(path=path_to_your_dataset)
stream = dataset.load_stream(stream_name="rsu_data", version=1, exact=True)

selected_columns = [
    "uuid",
    "dtc_vehicle_timestamp",
    "dtc_360_id",
    "model_range",
]
filter_expr = pc.field("model_range") == "G05"

result_generator = stream.read_data_batch_stream(
    batch_size=10,
    columns=selected_columns,
    filter_expr=filter_expr,
    output="pandas",
)

result_iterator = iter(result_generator)
try:
    first_batch = next(result_iterator).df
    print(first_batch.head())
except StopIteration:
    print("No data returned.")

Stream Data Access via Configuration File​

Streams can also be loaded via a YAML configuration file. This approach is useful for sharing configurations or running in production. The YAML file encodes the dataset, stream, and version to load. Use get_data() to load data from the config file.

All parameters available to read_data_stream() such as version, output, columns, and filter_expr also apply here. A practical advantage is that the YAML file can be version-controlled with Git so you can trace exactly which data was used for training.

Split Your Data: Filter Expression Split​

The following example shows a single-stream filter_expr split loaded via get_data(config=...).

from pathlib import Path
import yaml
from AIdataschema import AIdataset

path = Path.home()
path_to_your_dataset = path / "AIdata_manual"

yaml_str = """
datasetpath: {dataset_path}
streams:
  streamname: "rsu_data"
  version: 1
  output: "pandas"
  columns: ['uuid', 'dtc_vehicle_timestamp', 'dtc_360_id', 'model_range']
  data:
    - g05:
        - filter_expr: "model_range == 'G05' and dtc_360_id != 'None'"
""".format(dataset_path=path_to_your_dataset)

config = yaml.safe_load(yaml_str)
g05 = AIdataset.get_data(config=config)

g05.df.head()

Split Your Data: Proportion Split​

For machine learning training, it is standard to partition data into train, test, and validation sets. AI Dataschema supports creating multiple splits via the relative option. If the data is split into multiple tables, get_data() returns multiple objects, each representing one split.

import yaml
from AIdataschema import AIdataset

yaml_str = """
datasetpath: {dataset_path}
streams:
  streamname: "rsu_data"
  version: 1
  output: "pandas"
  columns: ['uuid', 'dtc_vehicle_timestamp', 'dtc_360_id', 'model_range']
  data:
    - train:
        relative:
          split_size: 0.7
    - test:
        relative:
          split_size: 0.3
""".format(dataset_path=path_to_your_dataset)

config = yaml.safe_load(yaml_str)
train, test = AIdataset.get_data(config=config)

train.df.head()
test.df.head()

Split Your Data: SQL-like Split​

To enable flexible, on-the-fly data views, SQL expressions can be embedded in the YAML configuration, targeting the columns available in your stream. Use standard SQL operators such as = and != in the WHERE clause. == belongs to filter_expr, not to SQL.

import yaml
from AIdataschema import AIdataset

yaml_str = """
datasetpath: {dataset_path}
streams:
  streamname: "rsu_data"
  version: 1
  output: "pandas"
  data:
      - train:
          sql:
              query: "SELECT uuid, dtc_vehicle_timestamp, dtc_360_id, model_range from rsu_data where model_range = 'G05' and dtc_360_id is not NULL"
      - test:
          sql:
              query: "SELECT uuid, dtc_vehicle_timestamp, dtc_360_id, model_range from rsu_data where model_range != 'G05' and dtc_360_id is not NULL"
""".format(dataset_path=path_to_your_dataset)

config = yaml.safe_load(yaml_str)
train, test = AIdataset.get_data(config=config)

train.df.head()
test.df.head()

Next Steps​