从结构化来源提取数据

使用 Python 的 ETL 和 ELT

Jake Roach

Data Engineer

源系统

本课程涵盖：

CSV 文件
Parquet 文件
JSON 文件
SQL 数据库

数据还来自：

API
数据湖
数仓
网页抓取
……以及更多！

读取 Parquet 文件

Parquet 文件：

开源的列式文件格式，优化字段存取效率
使用方式与 CSV 类似

import pandas as pd

# 读取 parquet 文件到内存
raw_stock_data = pd.read_parquet("raw_stock_data.parquet", engine="fastparquet")

¹ https://www.databricks.com/glossary/what-is-parquet

连接 SQL 数据库

可将 SQL 数据库中的数据拉取为 pandas DataFrame
需要连接 URI 用于创建引擎并连接数据库

import sqlalchemy
import pandas as pd

# 连接 URI: schema_identifier://username:password@host:port/db
connection_uri = "postgresql+psycopg2://repl:password@localhost:5432/market"
db_engine = sqlalchemy.create_engine(connection_uri)

# 查询 SQL 数据库
raw_stock_data = pd.read_sql("SELECT * FROM raw_stock_data LIMIT 10", db_engine)

模块化

将逻辑拆分为函数

提升管道可读性
遵循"不要重复自己"原则
加速排障

def extract_from_sql(connection_uri, query):
    # 创建引擎，查询数据并返回 DataFrame
    db_engine = sqlalchemy.create_engine(connection_uri)
    return pd.read_sql(query, db_engine)

extract_from_sql("postgresql+psycopg2://.../market", "SELECT ... LIMIT 10;")

让我们来练习！

使用 Python 的 ETL 和 ELT