Metaflow

Metaflow is an open-source, human-centric Python framework that helps data scientists and engineers build, run, and manage data-intensive workflows for data science, machine learning, and AI. Netflix originally developed Metaflow, and open sourced it in 2019. When you integrate Metaflow with Scality ARTESCA, Metaflow uses ARTESCA as its S3-compatible datastore for run snapshots, code packages, data artifacts, and metadata. This delivers reproducible and versioned ML pipelines on enterprise on-premises object storage.

Prerequisites

• An active Linux server with network access to the ARTESCA S3 endpoint.

Set Up Resources for Metaflow

1. Create an Account for Metaflow.

2. Create a bucket (e.g. metaflow) to store data and artifacts.

3. Create Identity and Access Management (IAM) users for Metaflow.

4. Create an IAM policy to grant Metaflow access to the S3 bucket.

   The following policy grants Metaflow least-privilege access to the bucket and the necessary bucket-level metadata operations.

   Note

   Replace each bucket_name placeholder in the policy with the actual name of your bucket.

   {
     "Version": "2012-10-17",
     "Statement": [
       {
         "Effect": "Allow",
         "Action": [
           "s3:PutObject",
           "s3:GetObject",
           "s3:DeleteObject",
           "s3:ListBucket",
           "s3:GetBucketLocation",
           "s3:GetBucketVersioning",
           "s3:GetBucketObjectLockConfiguration"
         ],
         "Resource": [
           "arn:aws:s3:::bucket_name",
           "arn:aws:s3:::bucket_name/*"
         ]
       }
     ]
   }
   

5. Create IAM User Access Keys.

Install Certificates

1. Download the CA Root certificate.

2. Import the Certificate.

Integrate Metaflow with ARTESCA

Apply these steps to a Linux server with access to the ARTESCA instance.

1. Create a Python virtual environment and install Metaflow.

   Note

   Refer to the Metaflow installation documentation for more information.

   python3 -m venv ~/metaflow-venv
   source ~/metaflow-venv/bin/activate
   pip install metaflow boto3
   

2. Create the Metaflow configuration directory.

   mkdir -p ~/.metaflow
   

3. Write the following content to ~/.metaflow/config.json.

   Note

   Replace the <YOUR_BUCKET_NAME_PLACEHOLDER>, <YOUR_ENDPOINT_PLACEHOLDER>, and <YOUR_CERTIFICATE_PATH_PLACEHOLDER> placeholders with the real values.

   {
     "METAFLOW_DEFAULT_DATASTORE": "s3",
     "METAFLOW_DATASTORE_SYSROOT_S3": "s3://<YOUR_BUCKET_NAME_PLACEHOLDER>/metaflow",
     "METAFLOW_S3_ENDPOINT_URL": "https://<YOUR_ENDPOINT_PLACEHOLDER>",
     "METAFLOW_S3_VERIFY_CERTIFICATE": "<YOUR_CERTIFICATE_PATH_PLACEHOLDER>/artesca-ca.pem",
     "METAFLOW_S3_ADDRESSING_STYLE": "path"
   }
   

4. Export the IAM credentials in the shell environment.

   Note

   Replace all five placeholders with the real values: <YOUR_ACCESS_KEY_PLACEHOLDER>, <YOUR_SECRET_KEY_PLACEHOLDER>, <YOUR_S3_ENDPOINT_URL_PLACEHOLDER>, <YOUR_S3_BUCKET_PLACEHOLDER>, and <YOUR_CA_BUNDLE_PATH_PLACEHOLDER>.

   export AWS_ACCESS_KEY_ID='<YOUR_ACCESS_KEY_PLACEHOLDER>'
   export AWS_SECRET_ACCESS_KEY='<YOUR_SECRET_KEY_PLACEHOLDER>'
   export AWS_S3_ENDPOINT_URL='<YOUR_S3_ENDPOINT_URL_PLACEHOLDER>'
   export AWS_S3_BUCKET='<YOUR_S3_BUCKET_PLACEHOLDER>'
   export AWS_CA_BUNDLE='<YOUR_CA_BUNDLE_PATH_PLACEHOLDER>'
   

5. Create a file named metaflow_artesca.py with the following content.

   This flow generates a 100 MiB sample dataset, uploads it to ARTESCA, downloads it back using the Metaflow S3 client, applies gzip compression, and uploads the optimized artifact and accompanying metadata. It uses a single Metaflow S3 integration path (no boto3 fallback logic).

   import gzip
   import hashlib
   import json
   import os
   import time
   
   from metaflow import FlowSpec, S3, current, step
   
   # --- Required environment variables (set these before running) ---
   # AWS_ACCESS_KEY_ID        - S3 access key
   # AWS_SECRET_ACCESS_KEY    - S3 secret key
   # AWS_S3_ENDPOINT_URL      - Artesca endpoint, e.g. https://s3.artesca.example.com
   # AWS_S3_BUCKET            - target bucket name
   # AWS_CA_BUNDLE            - path to artesca-ca.pem
   _REQUIRED_ENV_VARS = [
       'AWS_ACCESS_KEY_ID',
       'AWS_SECRET_ACCESS_KEY',
       'AWS_S3_ENDPOINT_URL',
       'AWS_S3_BUCKET',
       'AWS_CA_BUNDLE',
   ]
   
   def _validate_environment():
       missing = [v for v in _REQUIRED_ENV_VARS if not os.environ.get(v)]
       if missing:
           raise EnvironmentError(f"Missing required environment variables: {', '.join(missing)}")
   
       ca_bundle = os.environ['AWS_CA_BUNDLE']
       if not os.path.exists(ca_bundle):
           raise FileNotFoundError(f"SSL certificate not found: {ca_bundle}")
   
   
   def generate_large_dataset(size_in_mib: int) -> bytes:
       """Return a synthetic CSV dataset of approximately *size_in_mib* MiB as UTF-8 bytes."""
       print(f"Generating {size_in_mib} MiB dummy dataset...", flush=True)
   
       target_bytes = size_in_mib * 1024 * 1024
       categories = ['electronics', 'clothing', 'books', 'sports', 'home', 'automotive']
       regions = ['north', 'south', 'east', 'west', 'central']
       statuses = ['active', 'pending', 'completed', 'cancelled']
   
       lines = []
       current_size = 0
       record_id = 0
   
       while current_size < target_bytes:
           category = categories[record_id % len(categories)]
           region = regions[record_id % len(regions)]
           status = statuses[record_id % len(statuses)]
           price = 10.99 + (record_id % 1000) * 0.50
           quantity = 1 + (record_id % 100)
           line = (
               f"{record_id:08d},{category},{region},{price:.2f},{quantity},{status},"
               f"2024-07-{(record_id % 30) + 1:02d},"
               f"product_{record_id % 500:03d},description_for_product_{record_id % 200:03d}\n"
           )
           encoded = line.encode('utf-8')
           lines.append(encoded)
           current_size += len(encoded)
           record_id += 1
   
           if record_id % 50000 == 0:
               print(f"  Generated {current_size / (1024 * 1024):.1f} MiB ({record_id:,} records)...", flush=True)
   
       data = b''.join(lines)
       print(f"Dataset ready: {len(data) / (1024 * 1024):.2f} MiB, {record_id:,} records", flush=True)
       return data
   
   
   class ArtescaS3Dataset100MiB(FlowSpec):
       """Metaflow flow: generate, upload, compress, and verify a 100 MiB dataset on Artesca S3."""
   
       @step
       def start(self):
           """Validate environment, generate 100 MiB dataset, and upload to S3."""
           _validate_environment()
   
           dataset_bytes = generate_large_dataset(100)
           size_mib = len(dataset_bytes) / (1024 * 1024)
   
           self.original_data_hash = hashlib.sha256(dataset_bytes).hexdigest()
           self.original_data_size = len(dataset_bytes)
           self.dataset_size_mib = size_mib
   
           print(f"Dataset: {size_mib:.2f} MiB, SHA-256: {self.original_data_hash}", flush=True)
   
           s3_root = f"s3://{os.environ['AWS_S3_BUCKET']}/original-storage/{current.run_id}"
           key = "dummy_dataset_100mib.csv"
   
           t0 = time.time()
           with S3(s3root=s3_root) as s3:
               s3.put(key, dataset_bytes)
           elapsed = time.time() - t0
   
           self.original_s3_path = f"{s3_root}/{key}"
           print(f"Uploaded to {self.original_s3_path} in {elapsed:.1f}s ({size_mib * 8 / elapsed:.1f} Mbps)", flush=True)
   
           self.next(self.transform_and_optimize)
   
       @step
       def transform_and_optimize(self):
           """Download the dataset, verify integrity, compress, and upload the optimized copy."""
           print(f"Downloading {self.dataset_size_mib:.2f} MiB from {self.original_s3_path}", flush=True)
   
           t0 = time.time()
           with S3() as s3:
               dataset_bytes = s3.get(self.original_s3_path).blob
           print(f"Downloaded in {time.time() - t0:.1f}s", flush=True)
   
           actual_hash = hashlib.sha256(dataset_bytes).hexdigest()
           if actual_hash != self.original_data_hash:
               raise RuntimeError(
                   f"Integrity check failed: expected {self.original_data_hash}, got {actual_hash}"
               )
           print("Data integrity verified (SHA-256 match)", flush=True)
   
           lines_count = dataset_bytes.count(b'\n')
           print(f"Records: {lines_count:,}", flush=True)
   
           t0 = time.time()
           compressed = gzip.compress(dataset_bytes, compresslevel=6)
           compression_time = time.time() - t0
   
           original_mib = self.original_data_size / (1024 * 1024)
           compressed_mib = len(compressed) / (1024 * 1024)
           ratio_pct = (1 - len(compressed) / self.original_data_size) * 100
   
           print(
               f"Compressed {original_mib:.2f} MiB -> {compressed_mib:.2f} MiB "
               f"({ratio_pct:.1f}% reduction) in {compression_time:.1f}s",
               flush=True,
           )
   
           self.optimization_info = {
               'type': 'gzip_compression',
               'original_size_mib': round(original_mib, 2),
               'optimized_size_mib': round(compressed_mib, 2),
               'compression_ratio_percent': round(ratio_pct, 1),
               'compression_time_seconds': round(compression_time, 1),
               'compression_speed_mbps': round(original_mib * 8 / compression_time, 1),
               'records_processed': lines_count,
               'compression_level': 6,
               'timestamp': int(time.time()),
           }
   
           s3_root = f"s3://{os.environ['AWS_S3_BUCKET']}/optimized-storage/{current.run_id}"
           t0 = time.time()
           with S3(s3root=s3_root) as s3:
               s3.put("optimized_dataset_100mib.csv.gz", compressed)
               s3.put("optimization_metadata.json", json.dumps(self.optimization_info, indent=2).encode('utf-8'))
           elapsed = time.time() - t0
   
           self.optimized_s3_path = f"{s3_root}/optimized_dataset_100mib.csv.gz"
           self.metadata_s3_path = f"{s3_root}/optimization_metadata.json"
           print(f"Uploaded optimized dataset in {elapsed:.1f}s ({compressed_mib * 8 / elapsed:.1f} Mbps)", flush=True)
   
           self.next(self.report)
   
       @step
       def report(self):
           """Print a structured summary of the run."""
           info = self.optimization_info
           bucket = os.environ['AWS_S3_BUCKET']
           endpoint = os.environ['AWS_S3_ENDPOINT_URL']
           bandwidth_saved = info['original_size_mib'] - info['optimized_size_mib']
   
           print("\nS3 STORAGE OPTIMIZATION REPORT", flush=True)
           print("=" * 60, flush=True)
           print("\nOriginal dataset", flush=True)
           print(f"  Path:    {self.original_s3_path}", flush=True)
           print(f"  Size:    {info['original_size_mib']} MiB", flush=True)
           print(f"  Records: {info['records_processed']:,}", flush=True)
           print(f"  Hash:    {self.original_data_hash}", flush=True)
           print("\nOptimized dataset", flush=True)
           print(f"  Data:        {self.optimized_s3_path}", flush=True)
           print(f"  Metadata:    {self.metadata_s3_path}", flush=True)
           print(f"  Size:        {info['optimized_size_mib']} MiB", flush=True)
           print(f"  Reduction:   {info['compression_ratio_percent']}% ({bandwidth_saved:.1f} MiB saved)", flush=True)
           print("\nBucket layout", flush=True)
           print(f"  s3://{bucket}/original-storage/   -> source datasets", flush=True)
           print(f"  s3://{bucket}/optimized-storage/  -> compressed datasets", flush=True)
           print(f"\nEndpoint: {endpoint}", flush=True)
   
           self.next(self.end)
   
       @step
       def end(self):
           """Flow completion marker."""
           info = self.optimization_info
           print(
               f"\nDone: {info['original_size_mib']} MiB -> {info['optimized_size_mib']} MiB "
               f"({info['compression_ratio_percent']}% reduction, {info['records_processed']:,} records)",
               flush=True,
           )
   
   
   if __name__ == '__main__':
       ArtescaS3Dataset100MiB()
   

6. Run the flow.

   python metaflow_artesca.py run
   

   Metaflow validates the flow graph and executes each step. The output must contain:

   • The dataset upload (dummy_dataset_100mib.csv).

   • The optimized artifact upload (optimized_dataset_100mib.csv.gz).

   • The metadata upload (optimization_metadata.json).

   • An SHA-256 integrity check.

   • A final performance report.

Verify the Integration

1. Access the ARTESCA UI, and navigate to your S3 bucket.

2. Confirm the following objects exist under optimized-storage/<run_id>/ in the bucket (where <run_id> is printed in the flow output):

   • optimized_dataset_100mib.csv.gz

   • optimization_metadata.json