📚 课程目录第2章数据源探索

数据源探索：结构化、半结构化与非结构化

数据入湖的第一步，不是写代码，而是搞清楚你的数据长什么样。我见过太多团队，上来就搞 Flink 作业，结果发现源表字段类型都搞错了，白白浪费几天时间。这一章，我们就来聊聊怎么摸清数据源的底细。

核心观点：数据源的类型决定了入湖策略。结构化数据用 JDBC 同步，半结构化数据需要 Schema 推断，非结构化数据则要依赖元数据管理。搞混了，后面全是坑。

结构化数据源：MySQL 与 PostgreSQL

结构化数据，说白了就是有固定 Schema 的关系表。MySQL 和 PostgreSQL 是入湖场景里最常见的两个选手。我个人习惯先看表结构，再看数据量，最后决定用全量还是增量。

MySQL 入湖要点

MySQL 的入湖方式很成熟。JDBC 直连是最直接的办法，但要注意几个细节：

主键或时间戳：增量同步必须依赖这个字段。我建议用 updated_at 时间戳，比自增 ID 更可靠。
批量大小：一次拉取 10000 行比较稳妥。太大容易 OOM，太小网络开销受不了。
Binlog 捕获：实时场景下，Debezium + Kafka 是标配。但要注意 Binlog 格式必须设为 ROW。

我的经验：曾经有个项目，MySQL 表有 2 亿行，直接用 JDBC 全量拉取，跑了 6 个小时还没跑完。后来改成按日期分区 + 并行读取，40 分钟搞定。记住，分区键选对了，效率翻倍。

PostgreSQL 入湖要点

PostgreSQL 比 MySQL 多了一个好东西——逻辑复制。你可以直接用 pgoutput 插件捕获变更，不需要像 MySQL 那样依赖 Binlog。

特性	MySQL	PostgreSQL
增量捕获方式	Binlog（ROW 格式）	逻辑复制 / 时间戳
JDBC 驱动	mysql-connector-java	postgresql-42.x.x
分区支持	需手动分区	原生声明式分区
数据类型兼容	JSON 支持较弱	原生 JSONB，入湖更友好

嗯，这里要注意：PostgreSQL 的 TIMESTAMP 类型带时区信息，入湖到 Delta Lake 时最好统一转为 UTC。我吃过这个亏，数据对不上，排查了半天才发现是时区问题。

半结构化数据源：JSON 与 CSV

半结构化数据，说白了就是有结构但不固定。JSON 字段可以嵌套，CSV 列数可能不一致。这类数据入湖，核心挑战是 Schema 推断。

JSON 入湖实战

JSON 文件最常见的坑是嵌套层级太深。Spark 读取时默认会展开成多级列，但列名会变得很长，比如 address.city.zipcode。我个人习惯先做一次 Schema 探查：

# 用 Spark 探查 JSON Schema
df = spark.read.option("inferSchema", "true") \
    .option("multiLine", "true") \
    .json("s3://data-lake/raw/json/")

# 打印 Schema 结构
df.printSchema()

# 结果示例：
# root
#  |-- id: long (nullable = true)
#  |-- name: string (nullable = true)
#  |-- address: struct (nullable = true)
#  |    |-- city: string (nullable = true)
#  |    |-- zipcode: string (nullable = true)
#  |-- orders: array (nullable = true)
#  |    |-- element: struct (containsNull = true)
#  |    |    |-- order_id: long (nullable = true)
#  |    |    |-- amount: double (nullable = true)

避坑指南：我曾经遇到一个 JSON 文件，前 1000 行里某个字段全是 null，Spark 推断成 string 类型。结果第 1001 行突然冒出一个整数，作业直接报错。解决方案是：手动指定 Schema，或者用 samplingRatio=1.0 全量采样。

CSV 入湖实战

CSV 看起来简单，其实坑最多。列分隔符、引号转义、空值处理，每个细节都能让你抓狂。

分隔符：别默认逗号。我见过用竖线 | 的，也见过用制表符 \t 的。先看文件头。
引号：字段里如果包含逗号，必须用引号包裹。Spark 的 option("quote", "\"") 可以处理。
空值：CSV 里的空字符串和 null 是两回事。建议统一用 option("nullValue", "") 处理。

# CSV 入湖的标准配置
df = spark.read \
    .option("header", "true") \
    .option("delimiter", ",") \
    .option("quote", "\"") \
    .option("escape", "\"") \
    .option("nullValue", "") \
    .option("inferSchema", "true") \
    .csv("s3://data-lake/raw/csv/orders_2024.csv")

非结构化数据源：日志与图片

非结构化数据，说白了就是没有固定格式的二进制流。日志文件是文本，但每行格式可能不同；图片是二进制，没法直接查。这类数据入湖，重点不在数据本身，而在元数据。

日志文件入湖

日志文件通常按天或按小时滚动。我建议先做一次格式解析，提取出时间戳、日志级别、消息体等结构化字段，再入湖。原始日志文件可以存到对象存储，元数据写入 Lakehouse 表。

推荐做法：日志入湖 = 元数据表 + 原始文件路径。元数据表记录文件名、时间范围、行数、文件大小。查询时先过滤元数据，再按需读取原始文件。这样既保留了原始信息，又支持高效检索。

图片文件入湖

图片入湖更简单。直接把文件存到 S3 或 HDFS，路径信息写入 Lakehouse 表。如果需要做图像分析，可以额外提取 EXIF 信息、文件哈希、尺寸等元数据。

元数据字段	类型	说明
file_path	string	对象存储中的完整路径
file_name	string	原始文件名
file_size	long	文件大小（字节）
file_hash	string	MD5 或 SHA256 哈希
width	int	图片宽度（像素）
height	int	图片高度（像素）
exif_datetime	timestamp	拍摄时间

我的习惯：图片入湖时，我会额外生成缩略图（256x256），存到另一个目录。这样前端展示时不需要加载原图，节省带宽。缩略图的元数据也一并写入 Lakehouse 表。

数据源探查清单

最后，我整理了一份数据源探查清单。每次接新数据源，我都会按这个清单走一遍，避免遗漏：

数据源类型：结构化 / 半结构化 / 非结构化？
数据量级：全量多大？每天增量多少？
Schema 稳定性：字段会变吗？变更频率如何？
增量标识：有没有时间戳或自增 ID？
数据质量：空值比例？异常值？重复数据？
访问方式：JDBC / 文件 / API / 消息队列？
权限要求：是否需要加密传输？敏感字段脱敏？

嗯，这一章的内容就到这里。数据源探查是入湖的第一步，也是最容易被忽视的一步。把这一步做扎实了，后面的入湖作业才能跑得稳。记住，数据源摸得越清楚，入湖踩的坑越少。