Flat File Ingestion Patterns: Loading CSV, JSON, and Parquet from Local and Cloud Storage into Your Pipeline

Every data pipeline starts with a source. And despite the prevalence of databases, APIs, and streaming systems, a huge proportion of real-world data still lives in flat files. Finance teams export CSVs from accounting software. IoT devices emit JSON records. Data warehouses produce Parquet dumps for archival. Your job as a data engineer is to reliably pull that data into your pipeline — whether it's sitting on a local drive, an S3 bucket, or Google Cloud Storage.

The challenge isn't just "read the file." It's reading it correctly, handling edge cases gracefully, and building ingestion code that works just as well on the 50th run as it did on the first. The way you structure your ingestion layer has downstream consequences for everything — schema validation, performance, error recovery, and cost. Getting it right from the start saves you from painful rewrites later.

By the end of this lesson, you'll be able to write production-grade Python ingestion code for CSV, JSON, and Parquet files from both local filesystems and cloud object storage (S3 and GCS). You'll understand the tradeoffs of each format, recognize the common failure modes, and have patterns you can adapt for your own pipelines immediately.

What you'll learn:

• The core differences between CSV, JSON, and Parquet and when each format is appropriate for ingestion
• How to load each format from a local filesystem using Python and pandas/PyArrow
• How to connect to AWS S3 and Google Cloud Storage and read files without downloading them first
• How to handle encoding issues, malformed records, and schema mismatches
• How to build a reusable, format-agnostic ingestion layer that selects the right reader automatically

---

Prerequisites

• Basic Python familiarity (functions, loops, dictionaries)
• pip installed and able to create a virtual environment
• An AWS or GCP account is helpful for the cloud sections, but not required to follow the logic

Install the libraries we'll use throughout:

pip install pandas pyarrow boto3 google-cloud-storage fsspec s3fs gcsfs

---

Understanding the Three Formats Before You Write a Line of Code

Before we touch any files, you need to understand what you're actually dealing with. The three formats — CSV, JSON, and Parquet — aren't interchangeable. Each was designed for a different use case, and that design shapes how you ingest them.

CSV (Comma-Separated Values) is a plain text format. Every row is a line, every column is separated by a delimiter (usually a comma, sometimes a tab or pipe). It's universal — virtually every tool on earth can read and write it. But it carries zero type information. The number 42, the string "42", and the date "1970-01-11" are all just characters. Your ingestion code has to infer or enforce types after the fact.

JSON (JavaScript Object Notation) is a semi-structured format. Each record can be a nested object with arrays, sub-objects, and mixed types. This makes it ideal for API responses and event streams, but it creates a real challenge for ingestion: you often don't know exactly what shape a record will have until you read it. JSON files in data pipelines come in two flavors: a single large array [{...}, {...}], or newline-delimited JSON (NDJSON), where each line is a separate JSON object. The NDJSON format is much easier to stream and process incrementally.

Parquet is a binary columnar format. Unlike CSV and JSON, it stores data column-by-column rather than row-by-row, and it embeds full schema information including data types. It's compressed by default and designed specifically for analytical workloads. Reading only the columns you need is extremely efficient. The tradeoff: it's not human-readable, so debugging raw files requires tooling.

---

Loading CSV Files

Basic Local CSV Ingestion

Let's start with a realistic scenario: your company receives a daily sales report from a third-party vendor as a CSV file dropped into a shared folder. Here's what that file might look like:

order_id,customer_id,order_date,product_sku,quantity,unit_price,status
10001,C-4821,2024-03-15,SKU-7890,2,49.99,completed
10002,C-1193,2024-03-15,SKU-3344,1,129.00,pending
10003,C-9902,2024-03-16,SKU-7890,5,49.99,completed

Reading this with pandas is straightforward:

import pandas as pd

df = pd.read_csv("data/sales_report_2024-03-16.csv")
print(df.dtypes)
print(df.head())

But this naive approach has a problem: order_date will be loaded as a plain string, and order_id might be inferred as an integer when you actually want to treat it as an opaque identifier. Always specify your schema explicitly:

df = pd.read_csv(
    "data/sales_report_2024-03-16.csv",
    dtype={
        "order_id": str,
        "customer_id": str,
        "product_sku": str,
        "quantity": int,
        "unit_price": float,
        "status": str
    },
    parse_dates=["order_date"]
)

Handling Encoding and Delimiter Problems

The real world is messy. CSV files from legacy systems frequently arrive in Windows-1252 or Latin-1 encoding rather than UTF-8, causing exceptions when pandas tries to decode special characters in names, addresses, or product descriptions.

import chardet

def detect_encoding(filepath: str) -> str:
    with open(filepath, "rb") as f:
        raw = f.read(10000)  # Sample first 10KB
    result = chardet.detect(raw)
    return result["encoding"]

encoding = detect_encoding("data/vendor_products.csv")
df = pd.read_csv("data/vendor_products.csv", encoding=encoding)

For delimiter issues, use sep=None with engine='python' to let pandas sniff the separator:

df = pd.read_csv("data/report.txt", sep=None, engine="python")

Chunked Reading for Large CSV Files

When a CSV is larger than your available memory (multi-gigabyte files are common in production), don't try to load it all at once. Use pandas' chunking iterator:

def process_large_csv(filepath: str, chunk_size: int = 100_000):
    chunks = pd.read_csv(filepath, chunksize=chunk_size)
    
    processed_rows = 0
    for chunk in chunks:
        # Validate and transform each chunk
        chunk = chunk.dropna(subset=["order_id", "customer_id"])
        chunk["unit_price"] = chunk["unit_price"].clip(lower=0)
        
        # Write to your downstream target here
        # e.g., append to a database table
        processed_rows += len(chunk)
        print(f"Processed {processed_rows} rows so far...")
    
    return processed_rows

Each chunk is a regular DataFrame — you apply the same transformations, just in batches.

---

Loading JSON Files

Single-Object and Array JSON

A flat JSON file containing an array of records is common when you're working with API exports or webhook dumps:

import json
import pandas as pd

# A file containing: [{"event_id": "e1", ...}, {"event_id": "e2", ...}]
with open("data/events_2024-03-16.json", "r", encoding="utf-8") as f:
    records = json.load(f)

df = pd.DataFrame(records)

Newline-Delimited JSON (NDJSON)

For large event logs, NDJSON is far more common and practical. Pandas handles it directly:

# Each line is a separate JSON object:
# {"event_id": "e1", "ts": "2024-03-16T10:00:00Z", "user_id": "U-991", "action": "purchase"}
# {"event_id": "e2", "ts": "2024-03-16T10:01:14Z", "user_id": "U-445", "action": "view"}

df = pd.read_json("data/events_2024-03-16.ndjson", lines=True)

Flattening Nested JSON

Here's where JSON ingestion gets genuinely difficult. Real event data is often nested:

{
  "event_id": "e1",
  "user": {"id": "U-991", "country": "US"},
  "items": [
    {"sku": "SKU-7890", "qty": 2, "price": 49.99}
  ]
}

Use pd.json_normalize to flatten this into a tabular structure:

import json
import pandas as pd

with open("data/events_nested.json", "r") as f:
    records = [json.loads(line) for line in f]

df = pd.json_normalize(
    records,
    record_path=["items"],          # Expand the items array
    meta=["event_id", ["user", "id"], ["user", "country"]],  # Keep top-level fields
    meta_prefix="",
    errors="ignore"                 # Skip records missing expected fields
)

This produces one row per item, with event_id, user.id, and user.country replicated across rows. The errors="ignore" flag prevents a crash when a record is missing a field that meta expects.

---

Loading Parquet Files

Parquet is the most pleasant format to ingest because the schema is embedded in the file and types are preserved exactly. Loading locally is clean:

import pandas as pd

df = pd.read_parquet("data/transactions_2024-03-16.parquet")
print(df.dtypes)  # Types are preserved — no guessing required

For columnar selection (one of Parquet's biggest advantages), specify only the columns you need:

# We only need price data for a revenue calculation — skip everything else
df = pd.read_parquet(
    "data/transactions_2024-03-16.parquet",
    columns=["transaction_id", "order_date", "total_amount", "currency"]
)

For larger files or more control, use PyArrow directly:

import pyarrow.parquet as pq

table = pq.read_table(
    "data/transactions_2024-03-16.parquet",
    columns=["transaction_id", "order_date", "total_amount"]
)

# Convert to pandas only when you need DataFrame operations
df = table.to_pandas()

Reading Partitioned Parquet Datasets

In production, Parquet data is often stored as a partitioned dataset — a directory structure where each subdirectory represents a partition value, like year=2024/month=03/day=16/. PyArrow's dataset API handles this elegantly:

import pyarrow.dataset as ds

dataset = ds.dataset(
    "data/transactions/",
    format="parquet",
    partitioning="hive"  # Expects "key=value" directory naming
)

# Push filters down to the reader — only reads the relevant files
table = dataset.to_table(
    filter=ds.field("year") == 2024,
    columns=["transaction_id", "total_amount"]
)
df = table.to_pandas()

This is called predicate pushdown — the filter is applied at the storage layer before data is loaded into memory, which is enormously efficient.

---

Reading from Cloud Object Storage

Reading files from S3 or GCS isn't fundamentally different from reading local files — the key is using a filesystem abstraction layer that makes cloud storage look like a local path to your reading functions.

Reading from AWS S3

The s3fs library provides a filesystem interface for S3. Combined with fsspec, pandas and PyArrow can use S3 paths directly:

import pandas as pd
import s3fs

# Authentication is handled by boto3's credential chain:
# environment variables, ~/.aws/credentials, or IAM role
fs = s3fs.S3FileSystem()

# CSV from S3
df_csv = pd.read_csv("s3://my-data-bucket/raw/sales/2024-03-16/sales_report.csv")

# JSON from S3
df_json = pd.read_json(
    "s3://my-data-bucket/raw/events/2024-03-16/events.ndjson",
    lines=True
)

# Parquet from S3
df_parquet = pd.read_parquet(
    "s3://my-data-bucket/raw/transactions/2024-03-16/transactions.parquet"
)

When you need to list files in a bucket prefix (for example, to process all files in a date partition):

import s3fs

fs = s3fs.S3FileSystem()

files = fs.ls("my-data-bucket/raw/sales/2024-03-16/")
print(files)
# ['my-data-bucket/raw/sales/2024-03-16/sales_report_001.csv',
#  'my-data-bucket/raw/sales/2024-03-16/sales_report_002.csv']

dfs = []
for file_path in files:
    df = pd.read_csv(f"s3://{file_path}")
    dfs.append(df)

combined = pd.concat(dfs, ignore_index=True)

Reading from Google Cloud Storage

The pattern for GCS is nearly identical, using gcsfs:

import pandas as pd
import gcsfs

# Authenticate via Application Default Credentials
# Run: gcloud auth application-default login
fs = gcsfs.GCSFileSystem(project="my-gcp-project")

# Direct path syntax works with pandas
df_csv = pd.read_csv("gs://my-data-bucket/raw/sales/2024-03-16/sales_report.csv")
df_parquet = pd.read_parquet("gs://my-data-bucket/raw/transactions/transactions.parquet")

# List files in a GCS prefix
files = fs.ls("my-data-bucket/raw/sales/2024-03-16/")

For the broader context of connecting to various data sources including cloud storage, see the lesson on Connecting to Databases, Flat Files, and Cloud Storage for Data Pipeline Ingestion.

---

Building a Reusable Format-Agnostic Ingestion Layer

Pipelines that handle multiple file types end up with scattered, inconsistent ingestion logic if you're not careful. A better pattern is a single ingest_file function that detects the format and dispatches to the right reader:

import pandas as pd
from pathlib import Path
from typing import Optional, List

def ingest_file(
    filepath: str,
    columns: Optional[List[str]] = None,
    encoding: str = "utf-8",
    chunk_size: Optional[int] = None
) -> pd.DataFrame:
    """
    Load a CSV, JSON (or NDJSON), or Parquet file from a local path or
    cloud URL (s3://, gs://) into a pandas DataFrame.
    """
    # Normalize path to extract extension without query strings
    path = filepath.split("?")[0]  # Strip S3 presigned URL params if present
    suffix = Path(path).suffix.lower()

    if suffix == ".csv":
        if chunk_size:
            chunks = pd.read_csv(filepath, encoding=encoding, chunksize=chunk_size)
            return pd.concat(chunks, ignore_index=True)
        return pd.read_csv(filepath, encoding=encoding, usecols=columns)

    elif suffix in (".json", ".ndjson"):
        is_ndjson = suffix == ".ndjson"
        return pd.read_json(filepath, lines=is_ndjson)

    elif suffix in (".parquet", ".pq"):
        return pd.read_parquet(filepath, columns=columns)

    else:
        raise ValueError(
            f"Unsupported file format '{suffix}'. "
            f"Expected .csv, .json, .ndjson, or .parquet."
        )


# Usage examples:
df1 = ingest_file("data/sales_report.csv")
df2 = ingest_file("s3://my-bucket/events/events.ndjson")
df3 = ingest_file(
    "s3://my-bucket/transactions/transactions.parquet",
    columns=["transaction_id", "total_amount"]
)

This function is a foundation, not a final product. As your pipeline grows, you'll add retry logic, schema validation, logging, and environment-based configuration. The article on Building Your First Data Pipeline with Python shows how to assemble functions like this into a coherent end-to-end pipeline.

---

Hands-On Exercise

You're building a daily ingestion pipeline for a retail analytics team. They receive three files every morning, one from each of three regional offices:

• north_region_sales_2024-03-16.csv — standard CSV with order data
• east_region_events_2024-03-16.ndjson — NDJSON clickstream events
• west_region_transactions_2024-03-16.parquet — Parquet transaction dump

Your task:

1. Create sample files locally. For CSV, write 5-10 rows with columns: order_id, customer_id, order_date, product_sku, quantity, unit_price. For NDJSON, write 5-10 lines with: event_id, user_id, timestamp, action, product_sku. For Parquet, create a small DataFrame and save it with df.to_parquet("west_region_transactions_2024-03-16.parquet").

2. Use the ingest_file function from the section above to load all three files.

3. For each DataFrame, print the shape, column names, and dtypes.

4. Add a loaded_at column to each DataFrame containing the current UTC timestamp using pd.Timestamp.utcnow().

5. Challenge: Modify ingest_file to accept a date_columns parameter (a list of column names to parse as dates) and pass it through to pd.read_csv as parse_dates.

6. Cloud challenge: If you have S3 access, upload one of your sample files to a test bucket and verify you can read it back using the s3:// path syntax.

---

Common Mistakes & Troubleshooting

Mistake: Trusting pandas' type inference for CSV files

Pandas will guess column types, and it guesses wrong often enough to cause real problems. A column of mostly integers with one empty cell becomes float. A column of ZIP codes like "07030" loses its leading zero and becomes the integer 7030. Always define your dtype map for CSV ingestion in production.

Mistake: Loading multi-gigabyte files into memory on small machines

If you're running on a machine with 8GB of RAM and trying to load a 12GB CSV, you'll hit a memory error or cause your system to thrash. Use chunksize for CSV, or switch to Parquet with predicate pushdown for large analytical datasets.

Mistake: Assuming NDJSON when the file is actually array JSON

pd.read_json("file.json", lines=True) will fail with a confusing error if the file is actually a JSON array [{...}, {...}] rather than one-object-per-line. Always inspect the first few bytes of an unfamiliar JSON file before committing to a reading strategy.

Mistake: Ignoring the UserWarning: Columns (n) have mixed types

When pandas sees this warning, it means a column has inconsistent types — some rows are integers, some are strings. This usually indicates upstream data quality issues. Don't suppress the warning; investigate it. For pipelines where this matters, the Data Quality: Validation, Testing, and Monitoring Pipelines lesson covers how to build automated checks into your ingestion layer.

Mistake: Hardcoding cloud paths with no error handling

If a file doesn't exist in S3 or GCS, you'll get a FileNotFoundError (or a boto3-specific exception). In a production pipeline, a missing file might mean the upstream process failed — that's a situation that warrants alerting, not just crashing silently. Wrap cloud reads in try/except and integrate with your error handling and recovery strategy.

Mistake: Not accounting for schema drift over time

File formats from external vendors change. A CSV that had 12 columns last month might have 14 next month. Your pipeline will either silently drop columns it doesn't expect or crash trying to use a column that's been renamed. Building schema validation into your ingestion step — checking that expected columns are present and typed correctly — is essential. This problem gets deeper as your pipeline matures; the article on Schema Evolution Strategies for Production Data Pipelines covers the full range of approaches.

---

Summary & Next Steps

You've covered a lot of ground. You can now read CSV, JSON/NDJSON, and Parquet files from both local paths and cloud object storage (S3 and GCS) using pandas and PyArrow. You understand why each format behaves the way it does, how to handle the common failure modes, and how to build a reusable ingestion layer that you can drop into a real pipeline.

Here's a quick reference of the tools and approaches covered:

Format	Library	Key Parameters	Best For
CSV	pandas	dtype, parse_dates, chunksize, encoding	Universal compatibility
NDJSON	pandas	lines=True	Streaming event data
Nested JSON	pandas	json_normalize, record_path, meta	API exports
Parquet	pyarrow / pandas	columns, predicate pushdown	Analytics, large datasets
S3	s3fs + pandas	Direct s3:// paths	AWS-hosted data
GCS	gcsfs + pandas	Direct gs:// paths	GCP-hosted data

Where to go from here:

Ingestion is only the first stage of a pipeline. Once your data is loaded, you need to transform, validate, and deliver it somewhere useful. As you build out that pipeline, you'll want to understand how to load only the new records each run rather than re-reading the entire file every time — that's the topic of Incremental Loading Patterns: Timestamps, CDC, and Watermarks. You'll also eventually need to schedule your ingestion to run automatically on a defined cadence, which is covered in Scheduling and Orchestrating Pipelines with Airflow.

As your pipeline handles more files and more complexity, you'll also want to think about how to make your ingestion code part of a fully testable system where you can verify behavior confidently before deploying changes to production.