An object store can hold data between pipeline tasks and external processing systems. For a daily listings analysis, Airflow exports data from PostgreSQL to S3; a pandas job in a Docker container reads it, calculates price fluctuations, and writes results back. section 8.3.1
Local development can use MinIO as an S3 API–compatible object store. In the listings setup, Docker Compose runs PostgreSQL, MinIO, and Airflow together. This lets the pipeline read and write object-storage data without running AWS S3 locally. section 8.3.1
Preparing data for a model
A handwritten-digit pipeline copies an MNIST dataset to S3, then uses S3Hook to download it into an in-memory binary stream. A Python task unpacks and converts the data to the recordIO format expected by SageMaker’s K-Means model, then uploads the result to S3. Keeping the 15 MB input in memory avoids leftover local files; larger inputs call for a different memory-use decision. section 8.2.3 section 8.2
Listing 8.6 Transforming MNIST data to recordIO for SageMaker’s K-Means model
import gzip import io import pickle from airflow.operators.python import PythonOperator from airflow.providers.amazon.aws.hooks.s3 import S3Hook from sagemaker.amazon.common import write_numpy_to_dense_tensor def _extract_mnist_data(): s3hook = S3Hook() #1 # Download S3 dataset into memory mnist_buffer = io.BytesIO() mnist_obj = s3hook.get_key( #2 bucket_name="your-bucket", key="mnist.pkl.gz", ) mnist_obj.download_fileobj(mnist_buffer) # Unpack gzip file, extract dataset, convert, upload back to S3 mnist_buffer.seek(0) with gzip.GzipFile(fileobj=mnist_buffer, mode="rb") as f: #3 train_set, _, _ = pickle.loads(f.read(), encoding="latin1") output_buffer = io.BytesIO() write_numpy_to_dense_tensor( #4 file=output_buffer, array=train_set[0], labels=train_set[1], ) output_buffer.seek(0) s3hook.load_file_obj( #5 output_buffer, key="mnist_data", bucket_name="your-bucket", replace=True, ) extract_mnist_data = PythonOperator( task_id="extract_mnist_data", python_callable=_extract_mnist_data, )
The Python task shows S3Hook retrieving an object into BytesIO before decompression and conversion. section 8.2.3
Test an S3 task locally with `airflow tasks test` after configuring AWS credentials and initializing the Airflow database. These examples show S3-specific operators and hooks, plus an S3-compatible alternative; they do not demonstrate a storage-independent object-store API. section 8.2.3 section 8.3.1
