简介这份大数据入门指南面向零基础到进阶的学习者与开发者系统梳理Hadoop、Hive、Spark、Storm、Flink、HBase、Kafka、Zookeeper、Flume等主流组件的知识脉络帮助读者解决技术栈庞杂、环境搭建繁琐、命令与集群管理难以上手等问题。资源包共629个文件以380张png截图、101个md笔记、69个java源码、25个xml配置及scala、properties、json、parquet、orc等文件为主涵盖学习路线、思维导图、软件安装指南、环境搭建、命令实操、集群资源管理、分区、视图与数据查询等模块压缩包约20.75MB目录结构清晰便于按技术模块检索。目前已有155人学习下载。读者可借助图文笔记与示例代码快速搭建实验环境对照源码理解HDFS、HBase等组件的实际调用方式并利用思维导图与路线图规划学习路径适合作为大数据入门阶段的系统化参考资料。1. 从一堆组件名到一条数据链路这套大数据栈到底怎么串起来很多人第一次看到 Hadoop、Hive、Spark、Storm、Flink、HBase、Kafka、Zookeeper、Flume 这九个名字排在一起第一反应是打开搜索引擎逐个查查完更懵——每个组件都能单独写一本书但它们之间到底谁替代谁、谁依赖谁、先学哪个后学哪个没人给一句痛快话。我当年也是这么过来的在虚拟机里装了三天 Hadoop最后发现连一个像样的数据流都没跑通。这套栈真正要解决的问题只有一个数据从产生到产生价值中间要经过采集、缓冲、存储、计算、查询五个环节每个环节都有专门的组件负责它们不是竞争关系是流水线上的工位。Kafka 和 Flume 负责把数据接进来HDFS 和 HBase 负责存Spark 和 Flink 负责算Hive 负责用 SQL 查Zookeeper 负责协调Storm 是 Flink 出现之前流计算的过渡方案。这篇文章不逐个背组件定义而是按一条真实数据链路的搭建顺序把每个组件放在它该在的位置上告诉你最小可跑通的配置怎么写、参数怎么调、哪里最容易翻车。适合已经会 Linux 基本操作、想从零搭一套能跑通的大数据环境、或者面试前需要把知识串成线的人。2. 环境底座Hadoop 伪分布式与 Zookeeper 的第一次握手2.1 为什么伪分布式是唯一合理的起点完全分布式集群听起来专业但对入门者来说是灾难。三台虚拟机、SSH 互信、时间同步、防火墙规则任何一步出错都会让你在第一个小时就放弃。伪分布式把 NameNode、DataNode、ResourceManager、NodeManager 全部塞在一台机器上用不同端口区分虽然不能模拟真实负载但能让你把配置文件、启动顺序、日志排查这套流程完整走一遍。我一般建议用 4GB 内存以上的虚拟机CentOS 7 或 Ubuntu 20.04 都行JDK 选 8 或 11Hadoop 用 3.x 版本因为 2.x 的很多命令和 3.x 有差异网上搜到的教程混着看容易踩坑。安装前先确认三件事hostname能解析、ssh localhost免密能通、java -version有输出。这三步不过后面全是白费。# 配置免密登录伪分布式也必须做否则 start-dfs.sh 会卡住 ssh-keygen -t rsa -P -f ~/.ssh/id_rsa cat ~/.ssh/id_rsa.pub ~/.ssh/authorized_keys chmod 600 ~/.ssh/authorized_keys ssh localhost # 能直接进去说明免密成功这段命令的逻辑是生成密钥对并把公钥追加到授权文件-P 表示空密码避免启动脚本时交互输入。chmod 600是 SSH 的硬性要求权限不对会直接拒绝。验证方式就是ssh localhost不提示密码直接登录。接下来改core-site.xml、hdfs-site.xml、mapred-site.xml、yarn-site.xml四个文件。核心参数只有几个fs.defaultFS指向hdfs://localhost:9000dfs.replication设成 1伪分布式只有一个 DataNode设 3 会一直报副本不足mapreduce.framework.name设成 yarnyarn.nodemanager.aux-services设成 mapreduce_shuffle。改完执行hdfs namenode -format注意这个命令只能执行一次重复执行会导致 clusterID 不一致DataNode 起不来这是血泪经验。2.2 Zookeeper 在伪分布式里的角色与整合验证Zookeeper 不是 Hadoop 的必选项但后面 HBase、Kafka 都依赖它做协调。伪分布式环境下 Zookeeper 用单机模式跑就行改conf/zoo.cfg里的dataDir指向一个存在的目录clientPort保持 2181。启动命令是zkServer.sh start验证用zkCli.sh进去ls /能看到 zookeeper 节点就说明通了。Hadoop 和 Zookeeper 的整合主要体现在 HA 场景伪分布式用不上但你可以手动验证一下 Zookeeper 的写入读取为后面 Kafka 做准备# 启动 Zookeeper 后进入客户端 zkCli.sh -server localhost:2181 # 在客户端内执行 create /test_node hello_bigdata get /test_node # 输出 hello_bigdata 说明 Zookeeper 工作正常 delete /test_node quitcreate创建持久节点get读取内容delete清理。这一步看起来简单但很多人在 Kafka 启动时报Zookeeper connection refused根源就是 Zookeeper 根本没起来或者端口被占。启动前用netstat -tlnp | grep 2181确认端口空闲。提示Hadoop 和 Zookeeper 的日志默认在logs/目录下启动失败先看.out文件再看.log文件前者是标准输出后者是详细日志90% 的问题在前者就能定位。3. 数据管道Flume 采集到 Kafka 缓冲的完整配置3.1 Flume 的 Source-Channel-Sink 模型怎么落地Flume 的核心就三个词Source 负责收Channel 负责暂存Sink 负责发。入门最常用的组合是spooldir或taildir做 Sourcememory做 Channelkafka做 Sink。spooldir 监控一个目录有新文件就读取读完给文件加.COMPLETED后缀taildir 更实用可以监控多个文件并记录偏移量适合日志采集场景。配置文件flume-kafka.conf这样写# 定义 agent 名称 a1分别指定 source、channel、sink a1.sources r1 a1.channels c1 a1.sinks k1 # taildir source 配置监控 /data/logs 下的 .log 文件 a1.sources.r1.type TAILDIR a1.sources.r1.positionFile /data/flume/taildir_position.json a1.sources.r1.filegroups f1 a1.sources.r1.filegroups.f1 /data/logs/.*\.log a1.sources.r1.batchSize 100 # memory channel 配置容量 1000 事件事务容量 100 a1.channels.c1.type memory a1.channels.c1.capacity 1000 a1.channels.c1.transactionCapacity 100 # kafka sink 配置指向本地 kafka 的 topic a1.sinks.k1.type org.apache.flume.sink.kafka.KafkaSink a1.sinks.k1.kafka.bootstrap.servers localhost:9092 a1.sinks.k1.kafka.topic bigdata_topic a1.sinks.k1.kafka.flumeBatchSize 100 a1.sinks.k1.kafka.producer.acks 1 # 绑定 source-channel-sink a1.sources.r1.channels c1 a1.sinks.k1.channel c1positionFile是 taildir 的关键它记录每个文件读到哪一行Flume 重启后不会重复读也不会漏读。batchSize和flumeBatchSize控制吞吐太小会频繁发请求太大会增加延迟100 是入门比较稳的值。acks1表示 leader 写入就返回追求吞吐可以设 0追求可靠设 -1但入门阶段 1 够用。启动命令flume-ng agent -n a1 -c conf -f flume-kafka.conf -Dflume.root.loggerINFO,console。-n指定 agent 名-c指定配置目录-f指定配置文件最后的 logger 参数让日志打到控制台方便调试。3.2 Kafka 单机跑通与 Topic 操作Kafka 依赖 Zookeeper所以先确保 Zookeeper 在跑。Kafka 3.x 之后可以不用 Zookeeper 了但入门教程大多还是用 ZK 模式这里按 ZK 模式讲。改config/server.properties里broker.id0、listenersPLAINTEXT://localhost:9092、zookeeper.connectlocalhost:2181、log.dirs/data/kafka-logs。启动bin/kafka-server-start.sh -daemon config/server.properties-daemon让它后台跑。创建 topic 并验证# 创建 1 分区 1 副本的 topic bin/kafka-topics.sh --create --topic bigdata_topic \ --bootstrap-server localhost:9092 --partitions 1 --replication-factor 1 # 查看 topic 列表 bin/kafka-topics.sh --list --bootstrap-server localhost:9092 # 生产消息 bin/kafka-console-producer.sh --topic bigdata_topic --bootstrap-server localhost:9092 # 输入几行文本后 CtrlC 退出 # 消费消息从头开始读 bin/kafka-console-consumer.sh --topic bigdata_topic \ --bootstrap-server localhost:9092 --from-beginning--from-beginning是新手最容易忘的参数不加的话只能读到启动之后产生的消息会误以为数据没进去。分区数决定并行度单机入门设 1 就行设多了反而增加管理成本。副本数不能超过 broker 数量单机只能设 1设 2 会报错。Flume 和 Kafka 都跑起来后往/data/logs/扔一个.log文件然后在 Kafka 消费者终端应该能看到内容。这个链路通了说明采集和缓冲环节没问题可以往上叠计算层了。注意Flume 的 memory channel 在 agent 进程挂掉时数据会丢生产环境用 file channel但 file channel 慢很多。入门阶段用 memory 快速验证链路别纠结可靠性。4. 计算与查询Spark、Hive、Flink 的分工与最小跑通4.1 Spark 本地模式跑通第一个数据分析任务Spark 的安装比 Hadoop 简单下载解压后改spark-env.sh里的JAVA_HOME就行。入门先用local[*]模式不依赖 YARN启动快调试方便。spark-shell进去就能写 Scalapyspark进去写 Python我一般用 pyspark因为 Python 生态更顺手。一个典型的数据清洗任务读 JSON 文件过滤空值按字段分组统计。from pyspark.sql import SparkSession from pyspark.sql.functions import col, count, when # 创建 SparkSessionlocal[*] 表示用所有本地核心 spark SparkSession.builder \ .appName(DataCleanDemo) \ .master(local[*]) \ .config(spark.sql.shuffle.partitions, 4) \ .getOrCreate() # 读取 JSON 文件multiLine 允许跨行 JSON df spark.read.option(multiLine, true).json(/data/input/records.json) # 查看 schema 和前几行 df.printSchema() df.show(5, truncateFalse) # 过滤掉关键字段为空的行按 category 分组计数 cleaned df.filter(col(category).isNotNull() (col(amount) 0)) result cleaned.groupBy(category).agg(count(*).alias(cnt)) result.orderBy(col(cnt).desc()).show() spark.stop()spark.sql.shuffle.partitions默认是 200本地模式设成 4 能减少小文件和小任务开销。multiLine参数在 JSON 跨行时必须开否则解析出来全是 null。filter里用而不是and因为 DataFrame 的列表达式需要位运算符。truncateFalse让show不截断长字符串调试时很有用。Spark 读取 JSON 时如果字段类型推断错了可以手动传 schema比自动推断快很多也避免全量扫描。这个技巧在处理大文件时特别明显。4.2 Hive 的安装配置与小文件优化Hive 本质是把 SQL 翻译成 MapReduce 或 Spark 任务元数据存在 MySQL 或 Derby 里。入门用 Derby 就行但 Derby 不支持多会话稍微正式一点就换 MySQL。改hive-site.xml配置javax.jdo.option.ConnectionURL指向 MySQLhive.metastore.warehouse.dir指向 HDFS 路径。建表时最容易踩的坑是分隔符。默认分隔符是\001如果你用逗号分隔的 CSV必须显式指定CREATE TABLE IF NOT EXISTS user_behavior ( user_id STRING, item_id STRING, category STRING, behavior STRING, ts BIGINT ) ROW FORMAT DELIMITED FIELDS TERMINATED BY , STORED AS TEXTFILE; -- 加载本地数据到 Hive 表 LOAD DATA LOCAL INPATH /data/user_behavior.csv INTO TABLE user_behavior; -- 给每一行标号用 row_number 窗口函数 SELECT user_id, item_id, ROW_NUMBER() OVER (PARTITION BY user_id ORDER BY ts) AS rn FROM user_behavior LIMIT 10;ROW_NUMBER() OVER (PARTITION BY ... ORDER BY ...)是 Hive 窗口函数最常用的模式按用户分组、按时间排序、给每行一个序号。PARTITION BY决定分组维度ORDER BY决定组内排序两者缺一不可。小文件问题是 Hive 的经典痛点。每次INSERT都可能产生一堆小文件NameNode 内存被大量元数据占满。解决办法是在 SQL 前设置合并参数-- 在会话级别开启小文件合并 SET hive.merge.mapfiles true; SET hive.merge.mapredfiles true; SET hive.merge.size.per.task 134217728; -- 128MB SET hive.merge.smallfiles.avgsize 16777216; -- 16MB -- 或者用 DISTRIBUTE BY 控制 reducer 数量 INSERT OVERWRITE TABLE result_table SELECT * FROM source_table DISTRIBUTE BY CAST(rand() * 10 AS INT);hive.merge.size.per.task控制合并后每个文件的目标大小hive.merge.smallfiles.avgsize是触发合并的平均文件大小阈值。DISTRIBUTE BY强制走 reducer 并控制分区数比SET mapred.reduce.tasks更灵活。这些参数在网约车数据分析这类项目里几乎是必调的因为原始数据按天分区每天几百个小文件不合并查询会慢到怀疑人生。4.3 Flink 流处理与 JDBC 连接器异常排查Flink 和 Spark 的区别在于 Flink 是真正的流处理来一条处理一条Spark Streaming 是微批。入门用 Flink 的 DataStream API 写一个从 Kafka 读、写到 MySQL 的任务能覆盖大部分核心概念。// 核心逻辑Kafka source - 简单转换 - JDBC sink StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment(); env.enableCheckpointing(5000); // 每 5 秒做一次 checkpoint KafkaSourceString source KafkaSource.Stringbuilder() .setBootstrapServers(localhost:9092) .setTopics(bigdata_topic) .setGroupId(flink_group) .setStartingOffsets(OffsetsInitializer.latest()) .setValueOnlyDeserializer(new SimpleStringSchema()) .build(); DataStreamString stream env.fromSource(source, WatermarkStrategy.noWatermarks(), Kafka Source); // 这里做转换比如解析 JSON、过滤 DataStreamTuple2String, Integer parsed stream .map(new MapFunctionString, Tuple2String, Integer() { Override public Tuple2String, Integer map(String value) throws Exception { String[] parts value.split(,); return Tuple2.of(parts[0], Integer.parseInt(parts[1])); } }); // JDBC sink 写入 MySQL parsed.addSink(JdbcSink.sink( INSERT INTO result_table (name, cnt) VALUES (?, ?) ON DUPLICATE KEY UPDATE cnt ?, (ps, t) - { ps.setString(1, t.f0); ps.setInt(2, t.f1); ps.setInt(3, t.f1); }, JdbcExecutionOptions.builder() .withBatchSize(100) .withBatchIntervalMs(200) .withMaxRetries(3) .build(), new JdbcConnectionOptions.JdbcConnectionOptionsBuilder() .withUrl(jdbc:mysql://localhost:3306/bigdata) .withDriverName(com.mysql.cj.jdbc.Driver) .withUsername(root) .withPassword(password) .build() )); env.execute(Flink Kafka to MySQL);enableCheckpointing是 Flink 容错的核心间隔太短影响吞吐太长恢复时重复数据多5000 毫秒是入门常用值。OffsetsInitializer.latest()表示从最新消息开始读改成earliest()会从头读调试时用后者能看到历史数据。JDBC 连接器最常见的异常是No suitable driver found原因是 MySQL 驱动 jar 没放到lib/目录或者withDriverName写错了。另一个高频异常是Communications link failure通常是 MySQL 的wait_timeout到了连接被服务端断开解决办法是在 JDBC URL 后面加autoReconnecttrue或者调大wait_timeout。还有Duplicate entry报错说明主键冲突用ON DUPLICATE KEY UPDATE做幂等写入能解决。提示Flink 的 checkpoint 目录要配在flink-conf.yaml的state.checkpoints.dir默认是内存任务失败后状态全丢。入门至少配到本地文件系统生产环境配到 HDFS。5. 避坑与排查这套栈最容易翻车的五个地方5.1 坑一Hadoop 重复 format 导致 DataNode 消失现象是start-dfs.sh后jps看不到 DataNodeNameNode 日志报clusterID不一致。原因是执行了两次hdfs namenode -format每次 format 生成新的 clusterID而 DataNode 的 VERSION 文件里还是旧的。解决办法是删掉dfs.namenode.name.dir和dfs.datanode.data.dir指向的所有目录重新 format 一次然后只启动一次。如果数据不重要这是最快的恢复方式如果数据重要需要手动改 VERSION 文件里的 clusterID 对齐但入门阶段直接清空重来更省事。5.2 坑二Kafka 消费者收不到消息现象是生产者发了消息消费者终端一片空白。先检查--from-beginning有没有加没加只能读到启动后的消息。如果加了还是空检查 topic 名是否一致Kafka 的 topic 名大小写敏感。再检查bootstrap-server地址有的教程写localhost:9092有的写127.0.0.1:9092在容器环境里这两个可能解析到不同网卡。最后看消费者组的 offset如果之前用同一个 group.id 消费过offset 已经提交到最新位置加--from-beginning也没用需要换一个 group.id 或者用--reset-offsets重置。5.3 坑三Hive 查询报 Java heap space现象是执行一个简单的SELECT count(*)就报java.lang.OutOfMemoryError: Java heap space。原因是 Hive 默认的容器内存太小或者数据倾斜导致某个 reducer 处理了过多数据。解决办法分两步先调大容器内存SET mapreduce.map.memory.mb2048; SET mapreduce.reduce.memory.mb4096;再检查数据分布用GROUP BY的字段如果某个值占了 80% 以上数据就是数据倾斜需要加随机前缀打散或者用MAPJOIN处理小表关联。5.4 坑四Spark 任务卡在最后一个 stage现象是 Spark UI 上所有 stage 都完成了但任务就是不结束。常见原因是spark.sql.shuffle.partitions设得太大产生了大量空任务每个任务调度都有开销。本地模式设成 4 到 8 就够集群模式按核心数的 2 到 3 倍设。另一个原因是数据倾斜某个 partition 数据量是其他的几十倍其他任务秒完它跑几小时。解决办法是用salting技术给 key 加随机前缀或者用repartition重新分布。5.5 坑五Flink JDBC 连接器写入重复数据现象是 MySQL 里出现了重复记录任务重启后重复更严重。原因是 Flink 的 checkpoint 机制保证 at-least-once不是 exactly-once除非用两阶段提交。入门阶段最简单的解决办法是在 MySQL 表上建唯一索引写入时用INSERT ... ON DUPLICATE KEY UPDATE把重复写入变成幂等操作。另一个办法是调大 checkpoint 间隔减少重启次数但治标不治本。真正要 exactly-once 需要 JDBC 连接器支持两阶段提交配置withBatchSize和withBatchIntervalMs配合 checkpoint 使用但入门阶段用唯一索引就够了。6. 从能跑到好用一个验证链路是否健康的检查清单搭完这套环境怎么判断它是真的健康还是只是表面能跑我一般用下面这个清单过一遍每一项都对应一个具体命令或操作不靠感觉。检查项验证方式健康标准HDFS 读写hdfs dfs -put本地文件再-cat内容一致无报错YARN 资源yarn node -list节点状态 RUNNINGZookeeper 会话zkCli.sh创建临时节点后退出重进临时节点消失Kafka 吞吐生产者发 1000 条消费者计数1000 条全收到Flume 断点续传停 Flume追加日志重启新日志被采集旧日志不重复Hive 元数据show partitions和describe formatted分区和字段类型正确Spark 血缘df.explain(true)能看到逻辑计划和物理计划Flink checkpointWeb UI 的 Checkpoints 页面有 completed 记录大小稳定这个清单里最容易被忽略的是 Flume 断点续传和 Flink checkpoint。前者验证的是positionFile是否真的生效后者验证的是状态后端是否配置正确。这两个如果没配好任务跑起来看着正常一重启就出问题。还有一个进阶技巧用hadoop distcp做跨集群数据迁移时-m参数控制并发 map 数默认是 20小文件多的时候调到 50 能快很多但别超过集群的 map slot 总数。-bandwidth限制每个 map 的带宽单位是 MB/s在共享集群里限速能避免把网络打满。这些参数在面试里经常被问到但只有真正迁移过数据的人才知道什么时候该调。我自己的习惯是每搭完一个组件立刻写一个最小验证脚本存下来下次环境出问题先跑脚本能快速定位是哪个环节挂了。这套栈组件多、依赖复杂靠记忆排查不现实靠脚本和清单才是正经做法。希望帮到你。本文还有配套的精品资源点击获取
简介这份大数据入门指南面向零基础到进阶的学习者与开发者系统梳理Hadoop、Hive、Spark、Storm、Flink、HBase、Kafka、Zookeeper、Flume等主流组件的知识脉络帮助读者解决技术栈庞杂、环境搭建繁琐、命令与集群管理难以上手等问题。资源包共629个文件以380张png截图、101个md笔记、69个java源码、25个xml配置及scala、properties、json、parquet、orc等文件为主涵盖学习路线、思维导图、软件安装指南、环境搭建、命令实操、集群资源管理、分区、视图与数据查询等模块压缩包约20.75MB目录结构清晰便于按技术模块检索。目前已有155人学习下载。读者可借助图文笔记与示例代码快速搭建实验环境对照源码理解HDFS、HBase等组件的实际调用方式并利用思维导图与路线图规划学习路径适合作为大数据入门阶段的系统化参考资料。1. 从一堆组件名到一条数据链路这套大数据栈到底怎么串起来很多人第一次看到 Hadoop、Hive、Spark、Storm、Flink、HBase、Kafka、Zookeeper、Flume 这九个名字排在一起第一反应是打开搜索引擎逐个查查完更懵——每个组件都能单独写一本书但它们之间到底谁替代谁、谁依赖谁、先学哪个后学哪个没人给一句痛快话。我当年也是这么过来的在虚拟机里装了三天 Hadoop最后发现连一个像样的数据流都没跑通。这套栈真正要解决的问题只有一个数据从产生到产生价值中间要经过采集、缓冲、存储、计算、查询五个环节每个环节都有专门的组件负责它们不是竞争关系是流水线上的工位。Kafka 和 Flume 负责把数据接进来HDFS 和 HBase 负责存Spark 和 Flink 负责算Hive 负责用 SQL 查Zookeeper 负责协调Storm 是 Flink 出现之前流计算的过渡方案。这篇文章不逐个背组件定义而是按一条真实数据链路的搭建顺序把每个组件放在它该在的位置上告诉你最小可跑通的配置怎么写、参数怎么调、哪里最容易翻车。适合已经会 Linux 基本操作、想从零搭一套能跑通的大数据环境、或者面试前需要把知识串成线的人。2. 环境底座Hadoop 伪分布式与 Zookeeper 的第一次握手2.1 为什么伪分布式是唯一合理的起点完全分布式集群听起来专业但对入门者来说是灾难。三台虚拟机、SSH 互信、时间同步、防火墙规则任何一步出错都会让你在第一个小时就放弃。伪分布式把 NameNode、DataNode、ResourceManager、NodeManager 全部塞在一台机器上用不同端口区分虽然不能模拟真实负载但能让你把配置文件、启动顺序、日志排查这套流程完整走一遍。我一般建议用 4GB 内存以上的虚拟机CentOS 7 或 Ubuntu 20.04 都行JDK 选 8 或 11Hadoop 用 3.x 版本因为 2.x 的很多命令和 3.x 有差异网上搜到的教程混着看容易踩坑。安装前先确认三件事hostname能解析、ssh localhost免密能通、java -version有输出。这三步不过后面全是白费。# 配置免密登录伪分布式也必须做否则 start-dfs.sh 会卡住 ssh-keygen -t rsa -P -f ~/.ssh/id_rsa cat ~/.ssh/id_rsa.pub ~/.ssh/authorized_keys chmod 600 ~/.ssh/authorized_keys ssh localhost # 能直接进去说明免密成功这段命令的逻辑是生成密钥对并把公钥追加到授权文件-P 表示空密码避免启动脚本时交互输入。chmod 600是 SSH 的硬性要求权限不对会直接拒绝。验证方式就是ssh localhost不提示密码直接登录。接下来改core-site.xml、hdfs-site.xml、mapred-site.xml、yarn-site.xml四个文件。核心参数只有几个fs.defaultFS指向hdfs://localhost:9000dfs.replication设成 1伪分布式只有一个 DataNode设 3 会一直报副本不足mapreduce.framework.name设成 yarnyarn.nodemanager.aux-services设成 mapreduce_shuffle。改完执行hdfs namenode -format注意这个命令只能执行一次重复执行会导致 clusterID 不一致DataNode 起不来这是血泪经验。2.2 Zookeeper 在伪分布式里的角色与整合验证Zookeeper 不是 Hadoop 的必选项但后面 HBase、Kafka 都依赖它做协调。伪分布式环境下 Zookeeper 用单机模式跑就行改conf/zoo.cfg里的dataDir指向一个存在的目录clientPort保持 2181。启动命令是zkServer.sh start验证用zkCli.sh进去ls /能看到 zookeeper 节点就说明通了。Hadoop 和 Zookeeper 的整合主要体现在 HA 场景伪分布式用不上但你可以手动验证一下 Zookeeper 的写入读取为后面 Kafka 做准备# 启动 Zookeeper 后进入客户端 zkCli.sh -server localhost:2181 # 在客户端内执行 create /test_node hello_bigdata get /test_node # 输出 hello_bigdata 说明 Zookeeper 工作正常 delete /test_node quitcreate创建持久节点get读取内容delete清理。这一步看起来简单但很多人在 Kafka 启动时报Zookeeper connection refused根源就是 Zookeeper 根本没起来或者端口被占。启动前用netstat -tlnp | grep 2181确认端口空闲。提示Hadoop 和 Zookeeper 的日志默认在logs/目录下启动失败先看.out文件再看.log文件前者是标准输出后者是详细日志90% 的问题在前者就能定位。3. 数据管道Flume 采集到 Kafka 缓冲的完整配置3.1 Flume 的 Source-Channel-Sink 模型怎么落地Flume 的核心就三个词Source 负责收Channel 负责暂存Sink 负责发。入门最常用的组合是spooldir或taildir做 Sourcememory做 Channelkafka做 Sink。spooldir 监控一个目录有新文件就读取读完给文件加.COMPLETED后缀taildir 更实用可以监控多个文件并记录偏移量适合日志采集场景。配置文件flume-kafka.conf这样写# 定义 agent 名称 a1分别指定 source、channel、sink a1.sources r1 a1.channels c1 a1.sinks k1 # taildir source 配置监控 /data/logs 下的 .log 文件 a1.sources.r1.type TAILDIR a1.sources.r1.positionFile /data/flume/taildir_position.json a1.sources.r1.filegroups f1 a1.sources.r1.filegroups.f1 /data/logs/.*\.log a1.sources.r1.batchSize 100 # memory channel 配置容量 1000 事件事务容量 100 a1.channels.c1.type memory a1.channels.c1.capacity 1000 a1.channels.c1.transactionCapacity 100 # kafka sink 配置指向本地 kafka 的 topic a1.sinks.k1.type org.apache.flume.sink.kafka.KafkaSink a1.sinks.k1.kafka.bootstrap.servers localhost:9092 a1.sinks.k1.kafka.topic bigdata_topic a1.sinks.k1.kafka.flumeBatchSize 100 a1.sinks.k1.kafka.producer.acks 1 # 绑定 source-channel-sink a1.sources.r1.channels c1 a1.sinks.k1.channel c1positionFile是 taildir 的关键它记录每个文件读到哪一行Flume 重启后不会重复读也不会漏读。batchSize和flumeBatchSize控制吞吐太小会频繁发请求太大会增加延迟100 是入门比较稳的值。acks1表示 leader 写入就返回追求吞吐可以设 0追求可靠设 -1但入门阶段 1 够用。启动命令flume-ng agent -n a1 -c conf -f flume-kafka.conf -Dflume.root.loggerINFO,console。-n指定 agent 名-c指定配置目录-f指定配置文件最后的 logger 参数让日志打到控制台方便调试。3.2 Kafka 单机跑通与 Topic 操作Kafka 依赖 Zookeeper所以先确保 Zookeeper 在跑。Kafka 3.x 之后可以不用 Zookeeper 了但入门教程大多还是用 ZK 模式这里按 ZK 模式讲。改config/server.properties里broker.id0、listenersPLAINTEXT://localhost:9092、zookeeper.connectlocalhost:2181、log.dirs/data/kafka-logs。启动bin/kafka-server-start.sh -daemon config/server.properties-daemon让它后台跑。创建 topic 并验证# 创建 1 分区 1 副本的 topic bin/kafka-topics.sh --create --topic bigdata_topic \ --bootstrap-server localhost:9092 --partitions 1 --replication-factor 1 # 查看 topic 列表 bin/kafka-topics.sh --list --bootstrap-server localhost:9092 # 生产消息 bin/kafka-console-producer.sh --topic bigdata_topic --bootstrap-server localhost:9092 # 输入几行文本后 CtrlC 退出 # 消费消息从头开始读 bin/kafka-console-consumer.sh --topic bigdata_topic \ --bootstrap-server localhost:9092 --from-beginning--from-beginning是新手最容易忘的参数不加的话只能读到启动之后产生的消息会误以为数据没进去。分区数决定并行度单机入门设 1 就行设多了反而增加管理成本。副本数不能超过 broker 数量单机只能设 1设 2 会报错。Flume 和 Kafka 都跑起来后往/data/logs/扔一个.log文件然后在 Kafka 消费者终端应该能看到内容。这个链路通了说明采集和缓冲环节没问题可以往上叠计算层了。注意Flume 的 memory channel 在 agent 进程挂掉时数据会丢生产环境用 file channel但 file channel 慢很多。入门阶段用 memory 快速验证链路别纠结可靠性。4. 计算与查询Spark、Hive、Flink 的分工与最小跑通4.1 Spark 本地模式跑通第一个数据分析任务Spark 的安装比 Hadoop 简单下载解压后改spark-env.sh里的JAVA_HOME就行。入门先用local[*]模式不依赖 YARN启动快调试方便。spark-shell进去就能写 Scalapyspark进去写 Python我一般用 pyspark因为 Python 生态更顺手。一个典型的数据清洗任务读 JSON 文件过滤空值按字段分组统计。from pyspark.sql import SparkSession from pyspark.sql.functions import col, count, when # 创建 SparkSessionlocal[*] 表示用所有本地核心 spark SparkSession.builder \ .appName(DataCleanDemo) \ .master(local[*]) \ .config(spark.sql.shuffle.partitions, 4) \ .getOrCreate() # 读取 JSON 文件multiLine 允许跨行 JSON df spark.read.option(multiLine, true).json(/data/input/records.json) # 查看 schema 和前几行 df.printSchema() df.show(5, truncateFalse) # 过滤掉关键字段为空的行按 category 分组计数 cleaned df.filter(col(category).isNotNull() (col(amount) 0)) result cleaned.groupBy(category).agg(count(*).alias(cnt)) result.orderBy(col(cnt).desc()).show() spark.stop()spark.sql.shuffle.partitions默认是 200本地模式设成 4 能减少小文件和小任务开销。multiLine参数在 JSON 跨行时必须开否则解析出来全是 null。filter里用而不是and因为 DataFrame 的列表达式需要位运算符。truncateFalse让show不截断长字符串调试时很有用。Spark 读取 JSON 时如果字段类型推断错了可以手动传 schema比自动推断快很多也避免全量扫描。这个技巧在处理大文件时特别明显。4.2 Hive 的安装配置与小文件优化Hive 本质是把 SQL 翻译成 MapReduce 或 Spark 任务元数据存在 MySQL 或 Derby 里。入门用 Derby 就行但 Derby 不支持多会话稍微正式一点就换 MySQL。改hive-site.xml配置javax.jdo.option.ConnectionURL指向 MySQLhive.metastore.warehouse.dir指向 HDFS 路径。建表时最容易踩的坑是分隔符。默认分隔符是\001如果你用逗号分隔的 CSV必须显式指定CREATE TABLE IF NOT EXISTS user_behavior ( user_id STRING, item_id STRING, category STRING, behavior STRING, ts BIGINT ) ROW FORMAT DELIMITED FIELDS TERMINATED BY , STORED AS TEXTFILE; -- 加载本地数据到 Hive 表 LOAD DATA LOCAL INPATH /data/user_behavior.csv INTO TABLE user_behavior; -- 给每一行标号用 row_number 窗口函数 SELECT user_id, item_id, ROW_NUMBER() OVER (PARTITION BY user_id ORDER BY ts) AS rn FROM user_behavior LIMIT 10;ROW_NUMBER() OVER (PARTITION BY ... ORDER BY ...)是 Hive 窗口函数最常用的模式按用户分组、按时间排序、给每行一个序号。PARTITION BY决定分组维度ORDER BY决定组内排序两者缺一不可。小文件问题是 Hive 的经典痛点。每次INSERT都可能产生一堆小文件NameNode 内存被大量元数据占满。解决办法是在 SQL 前设置合并参数-- 在会话级别开启小文件合并 SET hive.merge.mapfiles true; SET hive.merge.mapredfiles true; SET hive.merge.size.per.task 134217728; -- 128MB SET hive.merge.smallfiles.avgsize 16777216; -- 16MB -- 或者用 DISTRIBUTE BY 控制 reducer 数量 INSERT OVERWRITE TABLE result_table SELECT * FROM source_table DISTRIBUTE BY CAST(rand() * 10 AS INT);hive.merge.size.per.task控制合并后每个文件的目标大小hive.merge.smallfiles.avgsize是触发合并的平均文件大小阈值。DISTRIBUTE BY强制走 reducer 并控制分区数比SET mapred.reduce.tasks更灵活。这些参数在网约车数据分析这类项目里几乎是必调的因为原始数据按天分区每天几百个小文件不合并查询会慢到怀疑人生。4.3 Flink 流处理与 JDBC 连接器异常排查Flink 和 Spark 的区别在于 Flink 是真正的流处理来一条处理一条Spark Streaming 是微批。入门用 Flink 的 DataStream API 写一个从 Kafka 读、写到 MySQL 的任务能覆盖大部分核心概念。// 核心逻辑Kafka source - 简单转换 - JDBC sink StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment(); env.enableCheckpointing(5000); // 每 5 秒做一次 checkpoint KafkaSourceString source KafkaSource.Stringbuilder() .setBootstrapServers(localhost:9092) .setTopics(bigdata_topic) .setGroupId(flink_group) .setStartingOffsets(OffsetsInitializer.latest()) .setValueOnlyDeserializer(new SimpleStringSchema()) .build(); DataStreamString stream env.fromSource(source, WatermarkStrategy.noWatermarks(), Kafka Source); // 这里做转换比如解析 JSON、过滤 DataStreamTuple2String, Integer parsed stream .map(new MapFunctionString, Tuple2String, Integer() { Override public Tuple2String, Integer map(String value) throws Exception { String[] parts value.split(,); return Tuple2.of(parts[0], Integer.parseInt(parts[1])); } }); // JDBC sink 写入 MySQL parsed.addSink(JdbcSink.sink( INSERT INTO result_table (name, cnt) VALUES (?, ?) ON DUPLICATE KEY UPDATE cnt ?, (ps, t) - { ps.setString(1, t.f0); ps.setInt(2, t.f1); ps.setInt(3, t.f1); }, JdbcExecutionOptions.builder() .withBatchSize(100) .withBatchIntervalMs(200) .withMaxRetries(3) .build(), new JdbcConnectionOptions.JdbcConnectionOptionsBuilder() .withUrl(jdbc:mysql://localhost:3306/bigdata) .withDriverName(com.mysql.cj.jdbc.Driver) .withUsername(root) .withPassword(password) .build() )); env.execute(Flink Kafka to MySQL);enableCheckpointing是 Flink 容错的核心间隔太短影响吞吐太长恢复时重复数据多5000 毫秒是入门常用值。OffsetsInitializer.latest()表示从最新消息开始读改成earliest()会从头读调试时用后者能看到历史数据。JDBC 连接器最常见的异常是No suitable driver found原因是 MySQL 驱动 jar 没放到lib/目录或者withDriverName写错了。另一个高频异常是Communications link failure通常是 MySQL 的wait_timeout到了连接被服务端断开解决办法是在 JDBC URL 后面加autoReconnecttrue或者调大wait_timeout。还有Duplicate entry报错说明主键冲突用ON DUPLICATE KEY UPDATE做幂等写入能解决。提示Flink 的 checkpoint 目录要配在flink-conf.yaml的state.checkpoints.dir默认是内存任务失败后状态全丢。入门至少配到本地文件系统生产环境配到 HDFS。5. 避坑与排查这套栈最容易翻车的五个地方5.1 坑一Hadoop 重复 format 导致 DataNode 消失现象是start-dfs.sh后jps看不到 DataNodeNameNode 日志报clusterID不一致。原因是执行了两次hdfs namenode -format每次 format 生成新的 clusterID而 DataNode 的 VERSION 文件里还是旧的。解决办法是删掉dfs.namenode.name.dir和dfs.datanode.data.dir指向的所有目录重新 format 一次然后只启动一次。如果数据不重要这是最快的恢复方式如果数据重要需要手动改 VERSION 文件里的 clusterID 对齐但入门阶段直接清空重来更省事。5.2 坑二Kafka 消费者收不到消息现象是生产者发了消息消费者终端一片空白。先检查--from-beginning有没有加没加只能读到启动后的消息。如果加了还是空检查 topic 名是否一致Kafka 的 topic 名大小写敏感。再检查bootstrap-server地址有的教程写localhost:9092有的写127.0.0.1:9092在容器环境里这两个可能解析到不同网卡。最后看消费者组的 offset如果之前用同一个 group.id 消费过offset 已经提交到最新位置加--from-beginning也没用需要换一个 group.id 或者用--reset-offsets重置。5.3 坑三Hive 查询报 Java heap space现象是执行一个简单的SELECT count(*)就报java.lang.OutOfMemoryError: Java heap space。原因是 Hive 默认的容器内存太小或者数据倾斜导致某个 reducer 处理了过多数据。解决办法分两步先调大容器内存SET mapreduce.map.memory.mb2048; SET mapreduce.reduce.memory.mb4096;再检查数据分布用GROUP BY的字段如果某个值占了 80% 以上数据就是数据倾斜需要加随机前缀打散或者用MAPJOIN处理小表关联。5.4 坑四Spark 任务卡在最后一个 stage现象是 Spark UI 上所有 stage 都完成了但任务就是不结束。常见原因是spark.sql.shuffle.partitions设得太大产生了大量空任务每个任务调度都有开销。本地模式设成 4 到 8 就够集群模式按核心数的 2 到 3 倍设。另一个原因是数据倾斜某个 partition 数据量是其他的几十倍其他任务秒完它跑几小时。解决办法是用salting技术给 key 加随机前缀或者用repartition重新分布。5.5 坑五Flink JDBC 连接器写入重复数据现象是 MySQL 里出现了重复记录任务重启后重复更严重。原因是 Flink 的 checkpoint 机制保证 at-least-once不是 exactly-once除非用两阶段提交。入门阶段最简单的解决办法是在 MySQL 表上建唯一索引写入时用INSERT ... ON DUPLICATE KEY UPDATE把重复写入变成幂等操作。另一个办法是调大 checkpoint 间隔减少重启次数但治标不治本。真正要 exactly-once 需要 JDBC 连接器支持两阶段提交配置withBatchSize和withBatchIntervalMs配合 checkpoint 使用但入门阶段用唯一索引就够了。6. 从能跑到好用一个验证链路是否健康的检查清单搭完这套环境怎么判断它是真的健康还是只是表面能跑我一般用下面这个清单过一遍每一项都对应一个具体命令或操作不靠感觉。检查项验证方式健康标准HDFS 读写hdfs dfs -put本地文件再-cat内容一致无报错YARN 资源yarn node -list节点状态 RUNNINGZookeeper 会话zkCli.sh创建临时节点后退出重进临时节点消失Kafka 吞吐生产者发 1000 条消费者计数1000 条全收到Flume 断点续传停 Flume追加日志重启新日志被采集旧日志不重复Hive 元数据show partitions和describe formatted分区和字段类型正确Spark 血缘df.explain(true)能看到逻辑计划和物理计划Flink checkpointWeb UI 的 Checkpoints 页面有 completed 记录大小稳定这个清单里最容易被忽略的是 Flume 断点续传和 Flink checkpoint。前者验证的是positionFile是否真的生效后者验证的是状态后端是否配置正确。这两个如果没配好任务跑起来看着正常一重启就出问题。还有一个进阶技巧用hadoop distcp做跨集群数据迁移时-m参数控制并发 map 数默认是 20小文件多的时候调到 50 能快很多但别超过集群的 map slot 总数。-bandwidth限制每个 map 的带宽单位是 MB/s在共享集群里限速能避免把网络打满。这些参数在面试里经常被问到但只有真正迁移过数据的人才知道什么时候该调。我自己的习惯是每搭完一个组件立刻写一个最小验证脚本存下来下次环境出问题先跑脚本能快速定位是哪个环节挂了。这套栈组件多、依赖复杂靠记忆排查不现实靠脚本和清单才是正经做法。希望帮到你。本文还有配套的精品资源点击获取
