Extracting Data from Flat Files and Databases

What Is Data Extraction?

Extraction is the first stage of every ETL pipeline. Its job is simple: read data from one or more source systems and move a faithful copy into a staging area without altering values. The staging area is a temporary holding zone—often a folder of flat files or a schema inside a database—where downstream transformation steps can work safely without touching the original source.

A well-written extraction script must do three things: connect to or open the source, read every row without dropping any, and verify the count before moving on. Row-count verification is not optional. If your staging file contains 9,847 rows but the source has 10,000, every transformation and load that follows will produce wrong answers silently.

Reading CSV Files with `csv.DictReader`

The Python standard library includes the `csv` module, which handles comma-separated value files reliably, including quoted fields that contain commas or newlines. The `csv.DictReader` class reads each row as a Python dictionary whose keys are the column headers from the first line of the file. This makes your code readable and independent of column order.

Basic pattern:

```python

import csv

rows = []

with open('sales.csv', newline='', encoding='utf-8') as infile:

reader = csv.DictReader(infile)

for row in reader:

rows.append(row)

print(f'Rows read: {len(rows)}')

```

Always open the file with `newline=''`—the csv module documentation requires this so that line endings inside quoted fields are handled correctly. Specify an encoding explicitly (`utf-8` is the safe default) to avoid errors on machines whose system locale differs from yours.

After reading, write the rows to a new staging CSV using `csv.DictWriter`, then compare `len(rows)` against the output file's row count. If they match, extraction is confirmed complete.

Querying a PostgreSQL Database with `psycopg2`

`psycopg2` is the standard Python adapter for PostgreSQL. It implements the DB-API 2.0 specification, which means its connection and cursor objects follow the same pattern used by other Python database drivers.

Connection pattern:

```python

import psycopg2

conn = psycopg2.connect(

host='192.168.1.10',

port=5432,

dbname='source_db',

user='etl_reader',

password='changeme'

)

cursor = conn.cursor()

cursor.execute("""

SELECT o.order_id, o.order_date, c.customer_name

FROM orders o

JOIN customers c ON o.customer_id = c.customer_id

""")

results = cursor.fetchall()

print(f'Rows fetched: {len(results)}')

cursor.close()

conn.close()

```

Use `fetchall()` for moderate result sets. For very large tables you would use `fetchmany(size)` in a loop to avoid loading millions of rows into memory at once, but for this lab the instructor-provisioned dataset is small enough for `fetchall()`.

Write results to a JSON-lines (`.jsonl`) staging file: one JSON object per line. This format is easy to parse downstream and works well with streaming readers.

Credential note: Never hard-code production passwords in a script you will share or submit. In real environments, credentials are stored in environment variables or a secrets manager. For this lab, use the credentials your instructor provides on the lab sheet only.

Calling a REST API with `requests`

Many modern data sources expose a REST API that returns JSON. The `requests` library (install with `pip install requests`) makes HTTP calls straightforward.

```python

import requests

response = requests.get('https://example.com/api/employees')

print(f'HTTP status: {response.status_code}')

data = response.json() # parse JSON body

print(f'Records received: {len(data)}')

```

Always check `response.status_code`. A `200` means success. Codes in the `4xx` range indicate a client error (wrong URL, missing authentication); codes in the `5xx` range indicate a server error. Log the status code before doing anything else—if it is not `200`, raise an exception or exit gracefully rather than writing empty or partial data to the staging area.

Append the parsed records to your staging area using `json.dumps()` per line in a `.jsonl` file.

Verifying Row Counts

Every extraction step must end with a printed log line that states the source name and the number of rows read. Compare that number to the instructor answer key. A mismatch of even one row is a failing extraction. This discipline carries directly into professional practice: data engineers and DBAs are accountable for data completeness from the moment they touch a source.

Safety and Professional Practice

Data extraction touches real or realistic databases. Observe these rules during all lab work:

• Read-only access only. Use SELECT statements. Never issue INSERT, UPDATE, DELETE, or DDL against the source database. Your lab account is provisioned as read-only, and the instructor can audit query logs.
• Protect credentials. Do not post connection strings, usernames, or passwords to shared drives, chat, or version-control repositories.
• Close connections. Always close cursors and connections in a `finally` block or use a context manager to ensure they are released even if an exception occurs. Unclosed connections consume server resources and can lock other users out.
• Log, do not silently swallow errors. Wrap network and database calls in `try/except` blocks that print a clear error message before the script exits. A silent failure is harder to debug than a noisy one.