Collecting and Storing Data for ML and AI on AWS (MLA-C02)

A SageMaker AI training job reads each input channel either from Amazon S3, using one of three input modes, or from a file system data source on Amazon EFS or Amazon FSx for Lustre. The choice decides how long a job waits before its first epoch, how much local storage it needs, and whether the training script has to change.

Data distribution across instances

For S3 channels, the S3DataDistributionType decides what each instance of a multi-instance job receives. FullyReplicated (the default) gives every instance the entire channel, which suits model-parallel jobs or algorithms that need all data on each node, but multiplies download size and local storage. ShardedByS3Key splits the objects so each instance gets a different subset, which suits data-parallel jobs where each worker should see only its own slice of the data. Input mode and distribution are independent settings, so changing one never forces a change to the other.

Amazon EBS appears in this task mainly as the training instance's ML storage volume: in File mode the volume must be large enough for the dataset, and a larger volume does not remove the per-job download time.

Use the managed export features, which read from backups or snapshots instead of the live database, so production serves no extra queries and consumes no extra capacity.

• DynamoDB export to Amazon S3 requires point-in-time recovery (PITR) to be enabled on the table. It reads from the continuous backup, so it consumes no read capacity and needs no code to page through items. Output is DynamoDB JSON or Amazon Ion in an S3 prefix. A full export captures the table at a point in time; an incremental export captures only items inserted, updated or deleted in a window you choose (from 15 minutes up to 24 hours), which is much cheaper than exporting a large table every day when only a small part changes. Exports have no built-in schedule; start them from a scheduler such as Amazon EventBridge Scheduler or a pipeline step.
• RDS and Aurora snapshot export to Amazon S3 extracts tables from an automated or manual DB snapshot (or an Aurora cluster) and writes them to S3 as Apache Parquet, ready for Athena, Glue or SageMaker. The running instance is not queried.

The general rule: any method that queries the live database, however it is scheduled, competes with production traffic. Change-capture features record only changes made after they are switched on and need consumer code, so on their own they cannot produce a complete historical copy. And a method whose output is another database, rather than files in S3, still leaves the export step undone.

Amazon OpenSearch Service is also a source: documents and logs indexed there can be read with the search APIs or snapshot tooling, but when it appears in Task 1.1 it is more often the destination, as a vector store (see below).

Pick the storage service by access pattern first, then the storage class by how often and how fast the data is read. Amazon S3 is the default home for training data of every type because it is durable, scales without limit and is readable in parallel by many jobs; file systems and block storage are chosen only when a workload needs their semantics or throughput.

S3 storage classes for ML data

A lifecycle rule is the right tool when you know when data cools; Intelligent-Tiering is the right tool when you don't. Intelligent-Tiering's optional Archive Access and Deep Archive Access tiers must be activated separately, so its default configuration never reaches Deep Archive pricing.

Configuring storage for compliance

Compliance for training data comes down to four controls: who holds the encryption keys, whether data can be deleted or altered, where it physically lives and is processed, and whether you know which data is sensitive.

• Key control and audit. SSE-KMS with a customer managed key lets the company own the key policy, rotate and disable the key, and see every encrypt and decrypt call in AWS CloudTrail. SSE-S3 uses keys that Amazon S3 manages, so it does not meet a "we control the keys" requirement.
• Training job encryption. A training job stores data in two places, each with its own key setting: KmsKeyId in OutputDataConfig encrypts the model artifacts written to S3, and VolumeKmsKeyId in ResourceConfig encrypts the ML storage volumes attached to the instances. Instance types that use only NVMe instance storage encrypt it with hardware-managed keys and do not take a volume key. Keys configured on other resources, and in-transit encryption settings, do not encrypt a training job's own storage.
• Immutability. S3 Object Lock (which requires versioning) keeps object versions from being deleted or overwritten for a retention period. Compliance mode cannot be shortened or bypassed by any user, including the root user. Governance mode can be bypassed by users granted s3:BypassGovernanceRetention. Versioning alone keeps old versions but does not stop deletion.
• Data residency. Data is processed where the job runs, and a SageMaker AI training job reads its dataset from the same Region as the job. To train in another Region you copy the data there first (for example with S3 Replication), and a residency rule limits which Regions are allowed for both the copy and the job. Residency is decided by Region choice alone; security controls protect data but do not change where it is stored or processed.
• Sensitive data discovery. Amazon Macie discovers personal and sensitive data in S3; automated sensitive data discovery samples objects continuously across buckets with managed data identifiers (personal identifiers, financial account details, credentials and more) and produces findings. Security services that watch access behaviour, resource configuration or software vulnerabilities do not look inside objects, so they cannot tell you where sensitive data lives.

Most capacity failures in ML ingestion come from load concentrated on one unit of capacity (a shard, a prefix, a consumer) or from too many tiny requests, not from total capacity. Find the bottleneck from the symptom before adding resources.

Kinesis capacity facts worth knowing: each shard accepts up to 1 MB per second or 1,000 records per second of writes and serves 2 MB per second of reads shared by standard consumers; on-demand mode scales shard count automatically but still hashes keys to shards.

Choose by what has to happen to the data in flight: Amazon Data Firehose when records only need delivering, Amazon Kinesis Data Streams when applications must read and replay them, Amazon MSK when producers speak Apache Kafka, and Amazon Managed Service for Apache Flink when the stream must be processed with state and time windows.

Consumers on Kinesis Data Streams

Standard (shared-throughput) consumers poll with GetRecords and split each shard's 2 MB per second of read throughput between them, with latency around 200 ms or more per consumer. Enhanced fan-out registers each consumer for its own dedicated 2 MB per second per shard, pushed over HTTP/2 with latency around 70 ms. Enhanced fan-out solves a throughput and latency problem between consumers; the retention period solves a replay problem. They are separate settings, so a requirement for both needs both. A message queue deletes messages once they are consumed, so it cannot stand in for a stream that several readers replay.

Configuring Firehose delivery for ML datasets

Firehose shapes what lands in S3 through four settings: record format conversion, data transformation with Lambda, dynamic partitioning, and buffering hints.

• Record format conversion writes Apache Parquet or ORC instead of the raw records, using a table in the AWS Glue Data Catalog as the schema. It accepts JSON input only (deserialized with the OpenX JSON SerDe or Hive JSON SerDe). CSV or other formats must first be turned into JSON by a Lambda transformation. A transformation function returns records, not files: Firehose concatenates whatever it returns into its own objects, so producing columnar files is the conversion feature's job.
• Data transformation invokes a Lambda function on each buffered batch; the function returns every record with its recordId, a result (Ok, Dropped or ProcessingFailed) and base64-encoded data. It changes record content, not delivery timing.
• Dynamic partitioning groups objects under S3 prefixes built from values inside each record, extracted with inline JSON parsing (jq expressions) or a Lambda function and referenced in the prefix with !{partitionKeyFromQuery:...}. Without dynamic partitioning, a prefix cannot use values from inside the record: its namespaces are !{timestamp:...}, which groups objects by arrival time, and !{firehose:...} (a random string, or the error type in an error output prefix). Partitioning on a high-cardinality key multiplies the number of objects.
• Buffering hints set a size (up to 128 MiB) and an interval (up to 900 seconds). Firehose writes an object when either limit is reached first. To make fewer, larger objects, raise the limit that is currently triggering the flush, then check that the other limit does not simply take over.

A worked example: a stream carrying about 300 MB per minute with hints of 64 MiB and 300 seconds reaches the size limit every 13 seconds or so, so the interval never matters. Lengthening the interval changes nothing; only a larger size hint (up to 128 MiB) produces bigger objects, and at that rate the interval still never fires.

Use a columnar format (Apache Parquet or ORC) for data that is written once and read by column for analytics and training, and a row-based format (CSV, JSON Lines, Avro) for data written or exchanged one record at a time or required by a specific consumer.

Format rules that trip people up

• SageMaker AI built-in algorithms with text/csv (for example XGBoost and Linear Learner) expect the target in the first column and no header row. A file with the target last trains a model that predicts whichever feature sits in the first column. Many built-in algorithms also accept RecordIO-protobuf, and XGBoost accepts LIBSVM and Parquet.
• Amazon Bedrock model customization takes training and validation data from S3 as JSON Lines: one JSON object per line, such as prompt and completion fields (or a messages structure for conversational models). The format is line-delimited, so every line must parse as a complete JSON object on its own.
• Schema enforcement on streams. AWS Glue Schema Registry stores schemas (Avro, JSON Schema or Protobuf) and applies compatibility modes such as BACKWARD, rejecting a producer's new schema version that would break consumers. Avro and Protobuf also give compact binary messages, while JSON Schema validates structure but keeps text payloads. Data without a registered schema gets no compatibility checks.

Pipelines usually combine both families: a row format where records are produced, and a columnar format, partitioned by the column most queries filter on, where they are analysed.

Merge large sources with a distributed Apache Spark engine, and pick where Spark runs by how much infrastructure the team will manage: AWS Glue runs serverless Spark jobs that read Data Catalog tables directly, including JDBC sources such as Amazon RDS, while Amazon EMR runs Spark on clusters you configure. Single-node tools are fine for small data but do not scale to terabytes.

Making a Glue join fast

• Prune partitions before reading. A push_down_predicate on a partition column (for example the date) filters partitions in the Data Catalog, so Glue never lists or reads the others, and it works for any date you rerun.
• Parallelise JDBC reads. Glue reads a JDBC table through one connection by default, which shows up as a single task. Setting hashfield (or hashexpression) with hashpartitions splits the read into parallel queries. Because the bottleneck is one connection, capacity added elsewhere in the job stays idle.
• Job bookmarks process only data added since the last run. They suit incremental pipelines; reprocessing a chosen historical date is a job for partition predicates.

Join strategy in Spark

Spark's default sort-merge join shuffles both tables by the join key, which is expensive when one side is terabytes. When the other side is small enough to fit in executor memory, a broadcast hash join sends the small table to every executor and avoids shuffling the large one. Spark broadcasts automatically only below spark.sql.autoBroadcastJoinThreshold (10 MB by default), so a larger reference table needs a broadcast hint or a higher threshold. Settings that tune the shuffle make it cheaper but still move the large table across the network; only a broadcast avoids that.

Amazon Athena can also join sources, including federated queries to databases through a Lambda-based connector, but the connector is something to deploy and a CTAS or INSERT INTO query can write at most 100 partitions, so Athena suits ad hoc merges more than large nightly builds.

Choose the vector store by query rate and latency, by whether vectors must sit beside relational data, by the retrieval features needed (such as hybrid keyword plus vector search), and by cost; then configure the index so its dimension, distance metric and algorithm match the embedding model and the workload.

Any store that holds a copy of operational data (OpenSearch, S3 Vectors) must be kept in sync, so its results lag committed changes; pgvector avoids that when the source is PostgreSQL. Amazon Bedrock Knowledge Bases also support other stores, such as Amazon Neptune Analytics and several partner databases.

Configuring the index to specification

• Dimension is fixed when the vector field or index is created and must equal the embedding model's output. Amazon Titan Text Embeddings V2 returns 1,024 dimensions by default (configurable to 512 or 256); the older Titan Embeddings G1 - Text returns 1,536. A mismatch makes every write fail; the fix is a new field or index (or a model whose output matches), because no other index setting changes its dimension.
• Distance metric (cosine, Euclidean/L2, inner product) should match how the model was trained; S3 Vectors supports cosine and Euclidean and fixes the metric at index creation.
• Algorithm. HNSW builds a graph that gives a strong speed-to-recall trade-off with no training step and can be built on an empty table; IVFFlat clusters vectors into lists, builds faster and uses less memory but needs representative data present before the index is built and usually recalls less.

Prerequisites for Amazon Bedrock knowledge base vector stores

• OpenSearch (Serverless or managed domain): create the index with a knn_vector field whose dimension matches the model and whose engine is faiss (the documentation states nmslib is not supported for managed domains), a text field for the chunk text, and the Bedrock-managed metadata field. Source details such as the S3 URI are kept in that metadata field.
• Aurora PostgreSQL with pgvector: the cluster must be in the same account as the knowledge base, with the RDS Data API enabled; create a table with an id (primary key), an embedding vector column, a chunks text column, a metadata column and optionally a custom_metadata column; create an HNSW index on the embedding with vector_cosine_ops, a GIN index on the chunks text (for hybrid search) and on custom_metadata if it is used for filtering; and store the database credentials in an AWS Secrets Manager secret.

Store unstructured objects (documents, images, audio, video) in Amazon S3 and keep their searchable attributes in a database or alongside the objects, with the S3 key as the link between them. S3 handles large objects, unlimited growth and highly parallel reads; databases are poor object stores (a DynamoDB item is limited to 400 KB, and relational BLOB columns are costly and slow at scale), and a single EBS volume is neither shared nor elastic.

• Metadata. Put per-object attributes that are queried by key (for example capture date, device, class label or owning team) in DynamoDB or a relational table, holding the S3 key. Manifest files list objects and their labels for a training channel or labeling job without moving data.
• Many small files. Millions of KB-sized images make training spend its time opening objects, even in FastFile mode. Pack samples into shard files of a few hundred MB, such as tar archives in WebDataset style, RecordIO or TFRecord, so loaders read sequentially. The fix has to reduce the number of objects; changes that leave the object count the same do not help.
• Documents for Amazon Bedrock Knowledge Bases. For an S3 data source, custom metadata comes from a sidecar file stored beside each document and named after it with the .metadata.json suffix (for example policy.pdf.metadata.json, holding a metadataAttributes object). Attributes are indexed with the chunks during sync and can be used as metadata filters at query time. For this data source the sidecar file is where ingestion reads custom metadata from; attributes kept anywhere else are not picked up. Chunking and parsing choices belong to Task 1.2.
• Audio and images feeding AI services. Amazon Transcribe, Amazon Textract and Amazon Rekognition read their inputs from S3 and can write outputs back to S3, so the same bucket layout serves both training and AI-service pipelines.

Ingest into a SageMaker Feature Store feature group with the PutRecord API for streaming, low-latency updates, and with batch ingestion (the Feature Store Spark connector, the SDK's FeatureGroup.ingest(), or a SageMaker Data Wrangler export) for bulk loads. Where the data lands depends on the feature group's online and offline store configuration.

A feature group can have either store or both; latest-value lookups and full history need both. Every record must include a record identifier feature and an event time feature. Because the online store keeps the newest event time, writing an older event time (for example when a backfill or a delayed producer writes an earlier timestamp) succeeds but does not overwrite a newer online value; the offline store keeps both records.

Choosing the ingestion path

• Streaming features (for example a Lambda function consuming Kinesis): call PutRecord per updated entity. The online store is updated at once and, with an offline store configured, the record is also replicated to S3. Records enter a feature group through the Feature Store APIs or connectors; files placed in the offline store's S3 location by other means are not feature group records.
• Backfills of billions of rows: run the Spark connector on Amazon EMR (or Glue/Processing with Spark) and set the target store. Targeting OfflineStore loads history for training without pushing stale values into the online store; single-process ingestion is meant for modest volumes, not billions of rows.
• Read-after-write in batch pipelines: offline replication of PutRecord data takes up to about 15 minutes, so SQL over the offline store reflects writes only after that delay. A pipeline that reads its own recent writes from the offline store has to allow for it (or read the latest values from the online store).

Store configuration that is fixed or easy to miss

• Offline table format is AWS Glue (default) or Apache Iceberg, chosen only at creation. With Iceberg, the many small Parquet files that frequent PutRecord calls produce can be compacted with Athena OPTIMIZE ... REWRITE DATA. To move an existing Glue-format feature group to Iceberg, create a new feature group and backfill it. Offline store partitioning is managed by Feature Store, so compaction, not re-partitioning, is the lever for small files.
• Time to live (TTL) applies to the online store only. Set TtlDuration as a feature-group default in the online store configuration, or per record in a PutRecord call; a record expires at its event time plus the duration. Expired records are hard-deleted from the online store, and the deletion is recorded in the offline store, so training history stays. A feature-group default applies to every record with no change to the code that writes them, and expiry needs no scheduled clean-up job. Training history survives because TTL never deletes from the offline store.

Building point-in-time correct training sets from the offline store, and managing feature definitions, are covered in the Task 1.2 lesson on feature engineering.