3. 环境准备：Flink on YARN/K8s集群搭建、Hadoop/Hive/MetaStore部署、MinIO/S3对象存储配置

说实话，很多做实时入湖的同学，最后翻车都不是翻在 Flink SQL 写得多花哨，而是翻在最基础的环境搭建上。我见过太多人把集群搭好了，结果跑了两天发现 Hive MetaStore 连不上，或者 S3 的权限策略配错了，数据写不进去。

这一章，我就把我在多个生产环境中沉淀下来的搭建经验，掰开了揉碎了讲给你听。咱们不搞那种「一键部署」的玩具脚本，而是让你真正理解每个组件为什么这么配。

3.1 对象存储：MinIO / S3 的选型与配置

对象存储是整个 Lakehouse 的底座。你想想看，所有的数据文件、Checkpoint、Savepoint 最终都要落到这里。选型上，我个人习惯分场景：

• 开发测试环境：用 MinIO。Docker 一键拉起，API 完全兼容 S3，成本几乎为零。
• 生产环境：用云厂商的 S3（AWS）或 OSS（阿里云）。别自己搭 MinIO 做生产，除非你们有专门的运维团队。

我记得有一次，一个客户在测试环境用 MinIO 跑得好好的，上了生产用 S3 就疯狂报错。查了半天，发现是 S3 的路径风格问题——MinIO 默认用 Path Style，而 AWS S3 强制要求 Virtual Hosted Style。这种坑，你提前知道就能省下半天时间。

MinIO 快速部署（Docker Compose）

version: '3.8'
services:
  minio:
    image: minio/minio:RELEASE.2024-01-01
    ports:
      - "9000:9000"   # API 端口
      - "9001:9001"   # Console 端口
    environment:
      MINIO_ROOT_USER: admin
      MINIO_ROOT_PASSWORD: admin123456
    volumes:
      - ./data:/data
    command: server /data --console-address ":9001"

启动之后，访问 http://localhost:9001，创建一个名为 lakehouse 的 Bucket。嗯，这里要注意：Bucket 名称必须全小写，而且不能有下划线，否则 Flink 写数据时会报 Illegal Bucket Name 的错误。

S3 客户端配置（AWS CLI）

aws configure
# 输入 Access Key / Secret Key
# 默认 region: us-east-1
# 输出格式: json

# 验证连通性
aws s3 ls s3://lakehouse/ --endpoint-url http://localhost:9000

3.2 元数据层：Hadoop / Hive / MetaStore 部署

元数据层是 Lakehouse 的「大脑」。Flink 写数据的时候，需要通过 Hive MetaStore 知道「这个表有哪些分区」「字段类型是什么」「数据存在哪个路径下」。

说白了，没有 MetaStore，你的数据就是一堆散落在 S3 上的 Parquet 文件，谁也查不了。

Hadoop 配置要点

Hadoop 在这里的角色主要是提供 core-site.xml 和 hdfs-site.xml。即使你不用 HDFS，Flink 和 Hive 也会读取这些配置来获取 S3 的认证信息。

我建议你直接在 core-site.xml 里配好 S3 的全局密钥：

<configuration>
  <property>
    <name>fs.s3a.access.key</name>
    <value>YOUR_ACCESS_KEY</value>
  </property>
  <property>
    <name>fs.s3a.secret.key</name>
    <value>YOUR_SECRET_KEY</value>
  </property>
  <property>
    <name>fs.s3a.endpoint</name>
    <value>http://localhost:9000</value>
  </property>
  <property>
    <name>fs.s3a.path.style.access</name>
    <value>true</value>
  </property>
</configuration>

Hive MetaStore 独立部署

很多教程把 Hive MetaStore 和 Hive Server2 混在一起讲。但在 Lakehouse 架构里，我们只需要 MetaStore，不需要 Hive 的计算能力。所以，独立部署 MetaStore 就够了。

关键配置项：

配置项	推荐值	说明
metastore.thrift.uris	thrift://localhost:9083	MetaStore 服务地址，Flink 通过这个端口连接
javax.jdo.option.ConnectionURL	jdbc:mysql://localhost:3306/hive_meta	元数据库，推荐 MySQL 8.0+，别用 Derby
metastore.warehouse.dir	s3a://lakehouse/warehouse	数据文件存储路径，直接指向 S3/MinIO
metastore.event.db.notification.api.auth	false	关闭通知认证，否则 Flink 写分区时会报权限错误

启动命令很简单：

nohup hive --service metastore > /var/log/hive-metastore.log 2>&1 &

验证是否启动成功：

netstat -tlnp | grep 9083
# 看到 LISTEN 状态就对了

3.3 计算引擎：Flink on YARN / K8s

终于到了 Flink 了。这里我重点讲两种部署模式的选择逻辑，而不是把官方文档抄一遍。

Flink on YARN：适合已有 Hadoop 集群的团队

如果你公司已经有现成的 YARN 集群，用 Flink on YARN 是最省事的。你不需要额外管理容器编排，YARN 帮你搞定资源分配和容错。

我常用的启动方式：

# 1. 设置 HADOOP_CLASSPATH，让 Flink 能找到 core-site.xml
export HADOOP_CLASSPATH=`hadoop classpath`

# 2. 启动一个 Flink 会话集群
./bin/yarn-session.sh \
  -jm 2048m \
  -tm 4096m \
  -s 4 \
  -nm flink-lakehouse

# 3. 提交作业到该会话
./bin/flink run \
  -c com.example.StreamingLakehouseJob \
  ./flink-lakehouse.jar

这里有个细节：-s 4 表示每个 TaskManager 启动 4 个 Slot。我建议 Slot 数不要超过 TaskManager 的 vCore 数，否则会出现线程争抢，反而拖慢性能。

Flink on K8s：适合云原生架构的团队

K8s 模式的好处是弹性好、部署快。但坑也不少。我刚开始用 Flink on K8s 的时候，被 flink-conf.yaml 里的 kubernetes.cluster-id 配置坑过——如果集群 ID 重复，Flink 会直接覆盖已有的集群，导致正在运行的作业被干掉。

推荐使用 Flink Native Kubernetes 模式：

./bin/flink run-application \
  -t kubernetes-application \
  -Dkubernetes.cluster-id=flink-lakehouse-${RANDOM} \
  -Dkubernetes.container.image=flink:1.18-scala_2.12 \
  -Dkubernetes.taskmanager.cpu=2 \
  -Dtaskmanager.memory.process.size=4096m \
  -Djobmanager.memory.process.size=2048m \
  -Dparallelism.default=4 \
  ./flink-lakehouse.jar

3.4 环境验证：跑一个简单的入湖作业

环境搭好了，怎么知道它能不能用？我一般会跑一个最简单的「心跳作业」来验证整条链路：

-- 在 Flink SQL CLI 中执行
CREATE CATALOG lakehouse WITH (
  'type' = 'hive',
  'hive-conf-dir' = '/etc/hive/conf',
  'default-database' = 'default'
);

CREATE TABLE sink_table (
  id INT,
  name STRING,
  ts TIMESTAMP(3)
) WITH (
  'connector' = 'filesystem',
  'path' = 's3a://lakehouse/warehouse/test_table',
  'format' = 'parquet'
);

INSERT INTO sink_table VALUES (1, 'hello lakehouse', CURRENT_TIMESTAMP);

如果这条 SQL 能跑通，并且你在 MinIO Console 里能看到新生成的 Parquet 文件，那恭喜你，环境准备阶段就圆满完成了。