Prepare Data

cats-dogs-train/
├── labels.csv
└── images/
    ├── cat1.jpeg
    ├── dog1.jpeg
    └── ...

# Training data
tracebloc data ingest ./cats-dogs-train \
  --task image_classification \
  --name cats_dogs_train \
  --intent train \
  --label-column label

# Test data
tracebloc data ingest ./cats-dogs-test \
  --task image_classification \
  --name cats_dogs_test \
  --intent test \
  --label-column label

tracebloc data list

helm repo add tracebloc https://tracebloc.github.io/client
helm repo update

apiVersion: tracebloc.io/v1
kind: IngestConfig
category: image_classification
table: cats_dogs_train
intent: train
csv: /data/shared/cats-dogs/labels.csv
images: /data/shared/cats-dogs/images/
label: label

# Train release — points at the ingest.yaml from step 3 (table: cats_dogs_train, intent: train)
helm install cats-dogs-train tracebloc/ingestor \
  --namespace <namespace> \
  --set-file ingestConfig=./ingest-train.yaml

# Test release — same shape, with table: cats_dogs_test and intent: test
helm install cats-dogs-test tracebloc/ingestor \
  --namespace <namespace> \
  --set-file ingestConfig=./ingest-test.yaml

ingestor = CSVIngestor(
    ...
    category=TaskCategory.TABULAR_CLASSIFICATION, # Adjust for your task
    csv_options=csv_options,      # Defined above
    label_column="ColumnName",    # Target column
    intent=Intent.TRAIN,          # TRAIN or TEST
)

# Build for cloud (multi-arch) and push directly to registry
docker buildx build --platform linux/amd64,linux/arm64 -t <your-username>/<image-name>:<tag> --push .

`kubectl apply -f ingestor-job.yaml -n <namespace>`

pip install tracebloc-ingestor

import logging

from tracebloc_ingestor import Config, Database, APIClient, CSVIngestor, run_ingestion
from tracebloc_ingestor.utils.logging import setup_logging
from tracebloc_ingestor.utils.constants import TaskCategory, Intent, DataFormat

config = Config()
setup_logging(config)
logger = logging.getLogger(__name__)


def main():
    # Initialize components
    database = Database(config)
    # Initialize API client
    api_client = APIClient(config)

    # Define csv_options and schema (schema is only needed for tabular data)
    csv_options = {...}
    schema = {...}

    # Initialize ingestor
    ingestor = CSVIngestor(...)

    # Run and ingest data
    run_ingestion(ingestor, config.LABEL_FILE, batch_size=config.BATCH_SIZE, logger=logger)


if __name__ == "__main__":
    main()

# Schema definition for tabular data
schema = {
  "feature_00": "FLOAT ",
  "feature_01": "FLOAT ",
  "feature_02": "FLOAT ",
  ...
}

# Image specific options including CSV options
image_options = {
    # Image processing options
    "target_size": (512, 512),  # (width, height) every image already has; nothing is resized
    "extension": FileExtension.JPG,  # allowed extension for images: jpeg, jpg, png
}

# Object detection specific options including CSV options
object_detection_options = {
    # Image processing options
    "target_size": (1920, 1080),  # (width, height) every image already has; nothing is resized
    "extension": FileExtension.JPG,  # allowed extension for images: jpeg, jpg, png
}

text_options = {"extension": FileExtension.TXT}  # Allowed text file extensions

csv_options = {
    "chunk_size": 1000,          # Process rows in batches for efficiency
    "delimiter": ",",            # Column separator
    "quotechar": '"',            # Quoted field character
    "escapechar": "\\",          # Escape character for quotes
    "encoding": "utf-8",         # File encoding
    "on_bad_lines": "warn",      # Log malformed rows instead of failing
    "skip_blank_lines": True,    # Ignore empty rows
    "na_values": ["", "NA", "NULL", "None"]  # Treat these as missing values
}

ingestor = CSVIngestor(
    database=database,                  # From ingestor-job.yaml
    api_client=api_client,              # From ingestor-job.yaml
    table_name=config.TABLE_NAME,       # From ingestor-job.yaml
    schema=schema,                      # Defined above, only needed for tabular data
    data_format=DataFormat.TABULAR,     # Set the data format for the task
    category=TaskCategory.TABULAR_CLASSIFICATION, # Adjust for your task
    csv_options=csv_options,            # Defined above
    file_options={"number_of_columns": len(schema)}, # Don´t change
    label_column="ColumnName",          # Target column
    intent=Intent.TRAIN,                # TRAIN or TEST
)

FROM python:3.11-slim
WORKDIR /app

RUN pip install --no-cache-dir tracebloc-ingestor

# Copy the ingestion script into /app
COPY ingestor.py /app/ingestor.py

# Set the entrypoint
ENTRYPOINT ["python", "/app/ingestor.py"]

RUN groupadd -g 1000 app && \
    useradd -u 1000 -g 1000 -m -s /bin/bash app && \
    chown -R 1000:1000 /app

USER 1000

# Set the entrypoint

docker buildx build --platform linux/amd64,linux/arm64 -t <your-username>/<image-name>:<tag> --push .

apiVersion: batch/v1
kind: Job
metadata:
  name: <JOBNAME> # Set a job name e.g. ingestor-job-train
  namespace: <namespace> # Use the client namespace
spec:
  template:
    spec:
      containers:
      - name: api
        image: <YOUR_DOCKER_USER>/<YOUR_IMAGE_NAME>:latest # Your Docker image name and tag, e.g. "latest"
        imagePullPolicy: Always  # Use IfNotPresent only for local tests
        # Required if the namespace enforces the `restricted` Pod Security Standard.
        # See "Run as non-root" below.
        securityContext:
          allowPrivilegeEscalation: false
          runAsNonRoot: true
          capabilities:
            drop:
              - "ALL"
          seccompProfile:
            type: RuntimeDefault
        volumeMounts:
          - name: shared-volume
            mountPath: "/data/shared" # Shared PVC of your secure environment — read your dataset from here
        env:
        # Client credentials
        - name: TRACEBLOC_ENV
          value: "prod"
        - name: CLIENT_ENV # older name of TRACEBLOC_ENV, for ingestor images that predate it; keep both equal
          value: "prod"
        - name: CLIENT_ID # Read from the Secret the installer created
          valueFrom:
            secretKeyRef:
              name: <namespace>-secrets
              key: CLIENT_ID
        - name: CLIENT_PASSWORD
          valueFrom:
            secretKeyRef:
              name: <namespace>-secrets
              key: CLIENT_PASSWORD

        # Storage configuration
        - name: CLIENT_PVC # value has to match the shared data PVC name in the client values.yaml
          value: "client-pvc"

        # MySQL configuration
        - name: MYSQL_HOST # value has to match the mysql deployment name in the client values.yaml
          value: "mysql-client"

        # Dataset information — paths inside the ingestor pod.
        # /data/shared is the mount of the client-pvc.
        - name: TRACEBLOC_SRC_PATH
          value: "/data/shared" # Root of the mounted dataset directory
        - name: TRACEBLOC_LABEL_FILE
          value: "/data/shared/labels.csv" # Path to the labels CSV inside the pod
        - name: TRACEBLOC_TABLE_NAME
          value: <UNIQUE_TABLE_NAME> # Different for train and test, no spaces
        - name: TRACEBLOC_TITLE
          value: <DATASET_TITLE> # Optional
        - name: TRACEBLOC_BATCH_SIZE
          value: "4000" # Optional, defaults to 4000
        - name: TRACEBLOC_LOG_LEVEL
          value: "DEBUG" # Set DEBUG, "WARNING", "INFO" or "ERROR"
      imagePullSecrets:
      - name: regcred
      volumes:
        - name: shared-volume
          persistentVolumeClaim:
            claimName: client-pvc # value has to match the shared data PVC name in the client values.yaml
      restartPolicy: Never

kubectl apply -f ingestor-job.yaml -n <namespace>
kubectl wait -n <namespace> --for=condition=complete job/<INGESTOR_JOB_NAME>
kubectl logs -n <namespace> job/<INGESTOR_JOB_NAME>

# Delete the job only after verifying logs
kubectl delete -n <namespace> job/<INGESTOR_JOB_NAME>

Warning: would violate PodSecurity "restricted:latest":
  allowPrivilegeEscalation != false (container "api" must set securityContext.allowPrivilegeEscalation=false),
  unrestricted capabilities (container "api" must set securityContext.capabilities.drop=["ALL"]),
  runAsNonRoot != true (pod or container "api" must set securityContext.runAsNonRoot=true),
  seccompProfile (pod or container "api" must set securityContext.seccompProfile.type to "RuntimeDefault" or "Localhost")
job.batch/ingestor-job-train-data created

securityContext:
  allowPrivilegeEscalation: false
  runAsNonRoot: true
  capabilities:
    drop:
      - "ALL"
  seccompProfile:
    type: RuntimeDefault

RUN groupadd -g 1000 app && \
    useradd -u 1000 -g 1000 -m -s /bin/bash app && \
    chown -R 1000:1000 /app

USER 1000

kubectl get jobs,pods -n <namespace>
kubectl logs -n <namespace> <pod-name>

kubectl delete job ingestor-job -n <namespace>
kubectl logs <pod-name>

kubectl get pvc -n <namespace>