📚 课程目录第3章 Flink与Hadoop生态集成

3. Flink与Hadoop生态集成：Flink读写HDFS、Flink与Hive集成、Flink on YARN/K8s部署

这一章，我们来聊聊Flink怎么跟Hadoop生态“搭伙过日子”。说白了，就是解决三个核心问题：数据怎么存、元数据怎么通、资源怎么管。我刚开始接触Flink时，也踩过不少坑，尤其是版本兼容性这块，简直让人头大。但搞懂了底层逻辑，你会发现这些集成其实很顺滑。

3.1 Flink读写HDFS：数据持久化的基石

HDFS是Lakehouse的底层存储基座。Flink读写HDFS，本质上是利用Hadoop的FileSystem API。你想想看，Flink作为一个流处理引擎，它的状态、Checkpoint、以及最终的Sink输出，都离不开可靠的持久化层。

3.1.1 核心依赖与配置

要在Flink中使用HDFS，首先得把Hadoop的依赖加进来。我个人习惯用Flink官方提供的“flink-shaded-hadoop”包，省去很多版本冲突的麻烦。

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-shaded-hadoop-2-uber</artifactId>
    <version>2.8.3-10.0</version>
</dependency>

配置上，你需要在flink-conf.yaml里设置好HDFS的默认文件系统：

fs.default-scheme: hdfs://namenode:8020

注意： 我曾经在生产环境遇到过一个问题——Flink任务一直报“FileSystem not found”。排查了半天，发现是Hadoop的core-site.xml和hdfs-site.xml没有放到Flink的classpath下。记得把这两个配置文件丢到$FLINK_HOME/conf目录下。

3.1.2 读写实战：StreamingFileSink与FileSource

Flink 1.12之后，StreamingFileSink成了写入HDFS的标准方式。它支持行编码和批量编码，还能按时间或文件大小滚动。

DataStream<String> input = ...;

StreamingFileSink<String> sink = StreamingFileSink
    .forRowFormat(new Path("hdfs:///user/flink/output"), 
                  new SimpleStringEncoder<String>("UTF-8"))
    .withBucketAssigner(new DateTimeBucketAssigner<>("yyyy-MM-dd--HH"))
    .withRollingPolicy(
        DefaultRollingPolicy.builder()
            .withRolloverInterval(TimeUnit.MINUTES.toMillis(15))
            .withInactivityInterval(TimeUnit.MINUTES.toMillis(5))
            .withMaxPartSize(1024 * 1024 * 128) // 128 MB
            .build())
    .build();

input.addSink(sink);

读取HDFS文件，我推荐用FileSource。它支持流式读取和批式读取，还能处理目录下的新文件。

FileSource<String> source = FileSource
    .forRecordStreamFormat(
        new TextLineInputFormat(),
        new Path("hdfs:///user/flink/input"))
    .monitorContinuously(Duration.ofSeconds(10)) // 流式模式，每10秒扫描一次
    .build();

DataStream<String> stream = env.fromSource(source, 
                                            WatermarkStrategy.noWatermarks(), 
                                            "hdfs-source");

我的经验： 在流式读取场景下，monitorContinuously的间隔不要设得太短。我曾经设成1秒，结果NameNode压力暴增。建议至少10秒以上，除非你的数据实时性要求极高。

3.2 Flink与Hive集成：打通批流元数据

Flink与Hive的集成，是实时湖仓一体落地的关键一步。它让Flink能直接读写Hive表，复用Hive的MetaStore和SQL语法。说白了，就是让Flink“说”Hive的话。

3.2.1 集成方式：HiveCatalog与HiveDialect

Flink提供了两种集成方式：

集成方式	核心作用	适用场景
HiveCatalog	将Hive MetaStore作为Flink的Catalog，统一元数据管理	需要复用Hive表结构、分区信息
HiveDialect	让Flink SQL解析Hive SQL语法	已有大量Hive SQL脚本，想平滑迁移

我个人更推荐优先使用HiveCatalog。它能让Flink和Hive共享同一套元数据，避免“表结构不一致”这种低级错误。

// 创建HiveCatalog
String name            = "myhive";
String defaultDatabase = "default";
String hiveConfDir     = "/etc/hive/conf";

HiveCatalog hive = new HiveCatalog(name, defaultDatabase, hiveConfDir);
tableEnv.registerCatalog("myhive", hive);
tableEnv.useCatalog("myhive");

3.2.2 读写Hive表实战

注册好Catalog后，你就可以像操作普通Flink表一样操作Hive表了。

// 读取Hive表
Table result = tableEnv.sqlQuery("SELECT * FROM myhive.ods.user_log WHERE dt = '2024-01-01'");

// 写入Hive表（动态分区写入）
tableEnv.executeSql("INSERT INTO myhive.dwd.user_event PARTITION (dt) " +
                     "SELECT user_id, event_type, event_time, dt FROM myhive.ods.user_log");

    关键点： Flink写入Hive表时，默认使用Parquet格式和Snappy压缩。如果你需要调整，可以在SQL中设置：SET 'table.exec.hive.fallback-mapred-reader' = 'true';。不过，我建议保持默认，性能最优。

3.2.3 避坑指南：分区与事务

嗯，这里要注意。Flink写入Hive表时，分区字段必须写在SQL的最后。而且，Hive的ACID事务表在Flink中支持有限。我曾经踩过一个坑：用Flink往Hive事务表写数据，结果一直报“Attempt to do update or delete using transaction manager that does not support these operations”。后来发现，Flink只支持Hive的非事务表。如果你需要事务支持，建议用Flink的Changelog-Stream模式配合Hive的Streaming ingest。

3.3 Flink on YARN/K8s：部署模式的抉择

部署Flink，说白了就是选“车”还是选“船”。YARN是传统大数据的老牌调度器，K8s是云原生的新贵。两者各有优劣，我根据项目经验总结了一张对比表。

维度	YARN	Kubernetes
资源粒度	Container（固定资源）	Pod（可动态调整）
弹性伸缩	依赖YARN队列，较慢	原生支持，秒级扩缩
运维复杂度	低（Hadoop集群自带）	高（需要维护K8s集群）
适用场景	传统Hadoop集群，离线/近实时	云原生环境，微服务架构

3.3.1 Flink on YARN：经典模式

YARN模式下，Flink有两种运行方式：Session模式和Per-Job模式。我个人更推荐Per-Job模式，因为它每个任务独立集群，资源隔离性好。

# Per-Job模式提交
./bin/flink run -m yarn-cluster \
    -yjm 1024 -ytm 2048 \
    -ys 2 \
    -p 4 \
    ./examples/streaming/WordCount.jar

参数说明：

-yjm：JobManager内存（MB）
-ytm：TaskManager内存（MB）
-ys：每个TaskManager的Slot数
-p：并行度

我的习惯： 在YARN上，我会把-ys设成与CPU核数一致。比如TaskManager是4核，我就设4个Slot。这样能最大化利用CPU资源，避免线程切换开销。

3.3.2 Flink on K8s：云原生之路

K8s部署Flink，我建议用Flink Operator。它把Flink集群抽象成K8s的CRD（自定义资源），管理起来非常方便。

# 部署Flink Operator
helm repo add flink-operator-repo https://downloads.apache.org/flink/flink-kubernetes-operator-1.7.0/
helm install flink-kubernetes-operator flink-operator-repo/flink-kubernetes-operator

# 提交Flink任务（通过CRD）
apiVersion: flink.apache.org/v1beta1
kind: FlinkDeployment
metadata:
  name: my-flink-job
spec:
  image: flink:1.17.0
  flinkVersion: v1_17
  flinkConfiguration:
    taskmanager.numberOfTaskSlots: "2"
  jobManager:
    resource:
      memory: "1024m"
      cpu: 1
  taskManager:
    resource:
      memory: "2048m"
      cpu: 2
  job:
    jarURI: local:///opt/flink/examples/streaming/WordCount.jar
    parallelism: 4
    upgradeMode: stateless

注意： K8s模式下，网络和存储是两大难点。我曾经遇到Flink任务频繁重启，原因是Pod间网络延迟过高。后来把Flink的taskmanager.network.memory.min从64MB调大到256MB，问题才解决。另外，Checkpoint的存储建议用PVC或S3，不要用本地磁盘，否则Pod重启后数据就丢了。

3.4 本章知识体系总览

为了让你更直观地理解本章的知识结构，我画了一张图。它展示了Flink与Hadoop生态集成的三个核心层次：存储、元数据、资源调度。

这张图清晰地展示了Flink与Hadoop生态集成的三个层次。从上到下，Flink通过HDFS、Hive和YARN/K8s，实现了从数据存储、元数据管理到资源调度的全链路打通。你想想看，这正是实时湖仓一体落地的技术基础。