Data Preparation and Pre-trained ML APIs

Why Data Preparation Matters

Data scientists spend 60-80% of their time on data preparation. The quality of your ML model is bounded by the quality of your data. Garbage in, garbage out.

On GCP, data preparation is not a single tool but a pipeline of services working together. Choosing the right tool for each stage is critical for cost, performance, and maintainability.

Pipeline Stages

Cloud Dataflow is Google Cloud's fully managed, serverless service for running Apache Beam pipelines. It handles both batch and streaming data processing with the same programming model.

Core Pipeline Concepts

Apache Beam Pipeline Example

# Apache Beam pipeline — runs on Dataflow (or locally for testing)
import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions

# Define pipeline options
options = PipelineOptions([
    '--runner=DataflowRunner',        # Use 'DirectRunner' for local testing
    '--project=my-gcp-project',
    '--region=us-central1',
    '--temp_location=gs://my-bucket/temp',
    '--staging_location=gs://my-bucket/staging',
])

# Build the pipeline
with beam.Pipeline(options=options) as p:
    (
        p
        | 'ReadFromBQ' >> beam.io.ReadFromBigQuery(
            query='SELECT * FROM `project.dataset.table`',
            use_standard_sql=True,
        )
        | 'FilterValid' >> beam.Filter(lambda row: row['amount'] > 0)
        | 'TransformData' >> beam.Map(
            lambda row: {
                'customer_id': row['customer_id'],
                'amount_normalized': row['amount'] / 1000.0,
                'category': row['category'].lower(),
            }
        )
        | 'WriteToBQ' >> beam.io.WriteToBigQuery(
            'project:dataset.output_table',
            schema='customer_id:STRING,amount_normalized:FLOAT,category:STRING',
            write_disposition=beam.io.BigQueryDisposition.WRITE_TRUNCATE,
        )
    )

# Streaming pipeline — process events from Pub/Sub in real time
with beam.Pipeline(options=options) as p:
    (
        p
        | 'ReadPubSub' >> beam.io.ReadFromPubSub(
            topic='projects/my-project/topics/events',
        )
        | 'ParseJSON' >> beam.Map(json.loads)
        | 'Window5Min' >> beam.WindowInto(
            beam.window.FixedWindows(300)  # 5-minute windows
        )
        | 'GroupByUser' >> beam.GroupByKey()
        | 'Aggregate' >> beam.Map(compute_user_stats)
        | 'WriteBQ' >> beam.io.WriteToBigQuery(...)
    )

Cloud Dataprep is a visual, no-code data wrangling tool built by Trifacta (now part of Alteryx). It provides an interactive web UI for exploring, cleaning, and transforming data without writing code.

When to Use Dataprep

• Visual data exploration — Dataprep shows data distributions, anomalies, and patterns automatically
• Non-technical users — Business analysts and data stewards who do not write Python or SQL
• One-off data cleaning — Ad-hoc data preparation tasks, not recurring production pipelines
• Data profiling — Quickly understand data quality (nulls, outliers, distributions)
• Recipe-based transforms — Build a sequence of transformation "recipes" that can be saved and reused

Under the hood, Dataprep generates and executes Dataflow jobs. So it is essentially a visual front-end for Dataflow with AI-assisted suggestions for transforms.

Cloud Dataproc is Google Cloud's managed service for running Apache Spark, Hadoop, Flink, and Presto clusters. It provisions clusters in 90 seconds and integrates with GCS, BigQuery, and other GCP services.

When to Use Dataproc

• Existing Spark/Hadoop code — Lift-and-shift from on-premises Hadoop clusters
• Spark MLlib — Distributed ML training using Spark's ML library
• Complex graph processing — GraphX, Spark GraphFrames
• PySpark/Scala Spark jobs — Teams that already know Spark
• Ephemeral clusters — Spin up a cluster for a job, shut it down when done (cost-effective)

# Create a Dataproc cluster via gcloud CLI
gcloud dataproc clusters create my-spark-cluster \
  --region=us-central1 \
  --master-machine-type=n1-standard-4 \
  --worker-machine-type=n1-standard-4 \
  --num-workers=2 \
  --image-version="2.1-debian11" \
  --enable-component-gateway \
  --optional-components=JUPYTER

# Submit a PySpark job
gcloud dataproc jobs submit pyspark \
  gs://my-bucket/scripts/etl_job.py \
  --cluster=my-spark-cluster \
  --region=us-central1 \
  -- --input=gs://my-bucket/raw/ --output=gs://my-bucket/processed/

# Delete cluster when done (ephemeral pattern)
gcloud dataproc clusters delete my-spark-cluster \
  --region=us-central1 --quiet

Comparison Table

Google Cloud offers pre-trained ML APIs that require zero training data and zero ML expertise. Send data via REST or client library, get structured predictions back. These are the fastest path to adding ML to an application.

Cloud Vision API

# Cloud Vision API — Label Detection (Python)
from google.cloud import vision

client = vision.ImageAnnotatorClient()

# From a GCS URI
image = vision.Image(source=vision.ImageSource(
    image_uri="gs://my-bucket/photo.jpg"
))

# Or from a URL
image = vision.Image(source=vision.ImageSource(
    image_uri="https://example.com/photo.jpg"
))

response = client.label_detection(image=image, max_results=10)

for label in response.label_annotations:
    print(f"{label.description}: {label.score:.2%}")

Cloud Natural Language API

# Cloud NLP API — Sentiment + Entity Analysis
from google.cloud import language_v1

client = language_v1.LanguageServiceClient()
document = language_v1.Document(
    content="Google Cloud's Vertex AI is an excellent ML platform.",
    type_=language_v1.Document.Type.PLAIN_TEXT,
)

# Sentiment
sentiment = client.analyze_sentiment(document=document).document_sentiment
print(f"Sentiment: score={sentiment.score:.2f}, magnitude={sentiment.magnitude:.2f}")

# Entities
entities = client.analyze_entities(document=document).entities
for entity in entities:
    print(f"  {entity.name}: {entity.type_.name} (salience={entity.salience:.2f})")

Cloud Speech-to-Text

Converts audio to text using Google's speech recognition models. Supports 120+ languages, streaming recognition (real-time), and batch transcription (long audio files).

• Synchronous — Audio up to 1 minute, immediate response
• Asynchronous — Audio up to 480 minutes, results via polling or callback
• Streaming — Real-time transcription as audio flows in

# Cloud Speech-to-Text — Transcribe audio from GCS
from google.cloud import speech

client = speech.SpeechClient()

audio = speech.RecognitionAudio(
    uri="gs://my-bucket/audio-sample.wav"
)
config = speech.RecognitionConfig(
    encoding=speech.RecognitionConfig.AudioEncoding.LINEAR16,
    sample_rate_hertz=16000,
    language_code="en-US",
    enable_automatic_punctuation=True,
    enable_word_time_offsets=True,
    diarization_config=speech.SpeakerDiarizationConfig(
        enable_speaker_diarization=True,
        min_speaker_count=2,
        max_speaker_count=4,
    ),
)

# For audio > 1 min, use long_running_recognize
operation = client.long_running_recognize(config=config, audio=audio)
response = operation.result(timeout=300)

for result in response.results:
    print(f"Transcript: {result.alternatives[0].transcript}")
    print(f"Confidence: {result.alternatives[0].confidence:.2%}")

Cloud Translation API

Two tiers of translation service:

# Cloud Translation API v3 (Advanced)
from google.cloud import translate_v2 as translate

client = translate.Client()

# Detect language
detection = client.detect_language("こんにちは世界")
print(f"Detected: {detection['language']} ({detection['confidence']:.0%})")

# Translate
result = client.translate(
    "Machine learning is transforming industries.",
    target_language="es",  # Spanish
)
print(f"Translation: {result['translatedText']}")

Video Intelligence API

Analyzes video content stored in Cloud Storage. Processes entire videos and returns time-stamped annotations.

# Video Intelligence API — Label Detection
from google.cloud import videointelligence

client = videointelligence.VideoIntelligenceServiceClient()

features = [videointelligence.Feature.LABEL_DETECTION]

operation = client.annotate_video(
    request={
        "input_uri": "gs://my-bucket/video.mp4",
        "features": features,
    }
)
result = operation.result(timeout=300)

for label in result.annotation_results[0].segment_label_annotations:
    print(f"Label: {label.entity.description}")
    for segment in label.segments:
        start = segment.segment.start_time_offset.seconds
        end = segment.segment.end_time_offset.seconds
        conf = segment.confidence
        print(f"  {start}s - {end}s (confidence: {conf:.2%})")

Authentication Methods

# Set ADC for local development
gcloud auth application-default login

# Set service account for production
export GOOGLE_APPLICATION_CREDENTIALS="/path/to/service-account-key.json"

# Enable an API
gcloud services enable vision.googleapis.com
gcloud services enable language.googleapis.com
gcloud services enable speech.googleapis.com
gcloud services enable translate.googleapis.com
gcloud services enable videointelligence.googleapis.com

Common Exam Scenarios