1. 为什么要在伪分布式集群里补上 Kafka 和 Sqoop 这两块拼图很多人在虚拟机或 Docker 里搭完 Hadoop、Hive、HBase 这一套之后会觉得集群已经跑起来了够用了。我一开始也是这么想的直到真正做数据链路的时候才发现少了 Kafka 和 Sqoop整个集群就是个信息孤岛——外部数据进不来实时数据流接不上只能靠手工传文件效率低得让人抓狂。先说 Sqoop 解决的是什么问题。我们平时做离线分析数据源大概率是关系型数据库比如 MySQL、PostgreSQL 这类。你不可能让业务方每天手动导出 CSV 再上传到 HDFS这既不现实也不可靠。Sqoop 的核心价值就是打通关系型数据库和Hadoop 生态之间的批量数据传输通道它底层其实是把导入导出任务翻译成 MapReduce 作业来跑利用集群的并行能力做高速搬运。没有它你的 Hive 数仓就是无源之水。再说 Kafka。伪分布式集群里跑批处理没问题但一旦涉及实时场景——比如日志采集、用户行为流、设备上报数据——你就需要一个高吞吐的消息中间件来缓冲和分发。Kafka 在伪分布式环境里的角色是充当数据总线上游生产者往里写下游 Spark Streaming、Flink 或者消费者程序从里面读。它和 HDFS 的区别在于HDFS 适合存大文件做离线分析Kafka 适合流式数据的即时传递。把这两个组件加到已有的伪分布式集群里本质上是在补全一条完整的数据链路关系型数据库 → Sqoop → HDFS/Hive → 计算引擎 → Kafka → 实时消费。这条链路打通之后你在这套环境上能做的实验和项目就完全不一样了从单纯的跑个 WordCount升级到模拟真实数据管道。需要提前说明的是伪分布式环境下 Kafka 和 Sqoop 的部署逻辑和真正的分布式集群有区别。Kafka 在伪分布式里通常就是单 Broker 模式ZooKeeper 可能复用已有的也可能是 Kafka 内置的Sqoop 则要注意它和 Hadoop 版本的兼容性尤其是 Hadoop 3.x 之后一些 API 的变化。下面我会按照实际操作的顺序把每一步的选择理由和踩坑点都讲清楚。2. 动手之前环境状态确认与版本兼容性排查2.1 先摸清现有集群的家底在加装任何组件之前我强烈建议你先花十分钟把现有环境的信息梳理清楚。我见过太多人直接上来就装结果 Sqoop 连不上 HDFS、Kafka 启动报 ZooKeeper 连接超时回头排查发现是版本对不上或者端口冲突。这些问题的根源都是没确认现状。你需要确认的信息包括这几项Hadoop 版本号执行hadoop version记下完整版本比如 3.3.4 还是 2.7.7。这个决定了 Sqoop 该选哪个版本。Java 版本java -versionSqoop 1.4.7 在 JDK 11 上会有反射相关的警告甚至报错JDK 8 最稳。ZooKeeper 是否已存在如果你之前装 HBase 时已经部署了 ZooKeeperKafka 可以直接复用如果没有Kafka 自带一个简易 ZooKeeper 可以用但生产环境不推荐。各组件端口占用情况netstat -tlnp | grep -E 2181|9092|8080确认 ZooKeeper 的 2181、Kafka 的 9092 没被占用。环境变量配置位置确认你的HADOOP_HOME、JAVA_HOME是在/etc/profile还是~/.bashrc里配的后面加 Kafka 和 Sqoop 要往同一个文件里加。把这些信息记在一个文本文件里后面配置的时候随时对照能省掉大量来回查的时间。2.2 版本搭配的坑Sqoop 和 Hadoop 3.x 的兼容问题这是我要重点提醒的一个坑。Sqoop 1.4.7 是 Apache 官方最后一个稳定发布版本它默认编译时依赖的是 Hadoop 2.x 的库。如果你现有集群是 Hadoop 3.x直接拿 Sqoop 1.4.7 去连大概率会遇到两类问题第一类是Guava 版本冲突。Hadoop 3.x 用的 Guava 版本比 Sqoop 1.4.7 自带的新启动时会报NoSuchMethodError或者ClassNotFoundException。解决办法是把$SQOOP_HOME/lib下的 guava jar 替换成和 Hadoop 一致的新版本具体版本号去$HADOOP_HOME/share/hadoop/common/lib里看。第二类是缺少 Hadoop 3 新增的依赖。Sqoop 运行时需要的一些 commons 库在 Hadoop 3 里路径或名称变了最典型的是commons-lang3。如果报错提示找不到某个类去 Hadoop 的 lib 目录里找到对应的 jar软链接或者复制到 Sqoop 的 lib 下。提示不要盲目地把 Hadoop lib 下所有 jar 都拷到 Sqoop lib 里这会导致更严重的冲突。正确做法是遇到一个报错补一个 jar每次只改一个变量方便定位。至于 Kafka版本选择相对宽松。Kafka 2.8 之后自带了 KRaft 模式不依赖 ZooKeeper但在伪分布式学习场景下我建议还是用带 ZooKeeper 的传统模式因为资料多、报错信息好查。Kafka 3.x 系列配合 ZooKeeper 3.5 都能跑选一个 3.4 或 3.6 的稳定版即可。注意 Kafka 的版本号和 Scala 版本是绑定的下载时文件名类似kafka_2.13-3.6.0.tgz2.13 是 Scala 版本选 2.12 或 2.13 都行看你集群里其他组件用的哪个。2.3 资源分配的现实考量伪分布式环境通常就跑在一台虚拟机或者一个 Docker 容器里内存是有限的。Hadoop 的 NameNode、DataNode、ResourceManager、NodeManager 加上 Hive 的 Metastore可能已经吃掉了 4-6GB 内存。再往上加 Kafka 和 Sqoop你得算一下账。Kafka 单 Broker 默认启动大概占 1GB 左右堆内存可以通过KAFKA_HEAP_OPTS调小到 512MB。Sqoop 本身不常驻只在执行任务时启动 MapReduce 作业所以它不占常驻内存但跑任务时会向 YARN 申请容器资源。如果你的 YARN 配置的可用内存本来就不多Sqoop 任务可能因为申请不到资源而卡在 ACCEPTED 状态。我的建议是在加装之前先看一眼 YARN 的可用资源打开 ResourceManager 的 Web UI默认 8088 端口看 Cluster Metrics 里的 Memory Total 和 Memory Used。如果剩余可用内存低于 2GB先把 YARN 的yarn.nodemanager.resource.memory-mb调大一些或者把 Kafka 的堆内存压到最低。3. Kafka 在伪分布式里的落地从单 Broker 配置到消息收发验证3.1 解压、环境变量与 broker.id 的设定Kafka 的安装本身很简单解压到/opt或者你习惯的目录下然后配环境变量。但有几个配置项在伪分布式场景下需要特别说明。先做基础操作tar -zxvf kafka_2.13-3.6.0.tgz -C /opt/ mv /opt/kafka_2.13-3.6.0 /opt/kafka然后在/etc/profile里加export KAFKA_HOME/opt/kafka export PATH$PATH:$KAFKA_HOME/binsource /etc/profile之后进入$KAFKA_HOME/config目录核心配置文件是server.properties。伪分布式单 Broker 模式下需要改动的关键项如下配置项默认值建议值说明broker.id00单 Broker 保持 0 即可集群中必须唯一listenersPLAINTEXT://:9092PLAINTEXT://:9092监听所有网卡方便宿主机访问log.dirs/tmp/kafka-logs/opt/kafka/data别用 /tmp重启会丢数据zookeeper.connectlocalhost:2181localhost:2181复用已有 ZooKeeper 或指向 Kafka 自带的num.partitions13学习分区机制时建议设 3方便观察offsets.topic.replication.factor31单 Broker 必须改成 1否则创建 topic 会失败这里重点说两个坑。第一个是log.dirs默认指向/tmp下的目录很多 Linux 发行版重启会清理/tmp你的 topic 数据就全没了。改成持久化目录是必须的。第二个是offsets.topic.replication.factor默认值是 3意思是 Kafka 内部用来存消费偏移量的 topic 要有 3 个副本。但你只有一个 Broker根本凑不齐 3 个副本结果就是创建任何 topic 都报错。这个坑非常隐蔽因为报错信息不会直接告诉你副本数超了而是提示 Error while creating topic 之类的模糊信息。改成 1 就好了。3.2 启动顺序与 ZooKeeper 的依赖关系Kafka 启动前必须先确保 ZooKeeper 在跑。如果你之前装 HBase 时已经启动了 ZooKeeper用zkServer.sh status确认一下状态是 leader 或 follower单机模式下通常是 standalone。如果没有可以用 Kafka 自带的# 启动 Kafka 自带的 ZooKeeper仅学习环境用 $KAFKA_HOME/bin/zookeeper-server-start.sh -daemon $KAFKA_HOME/config/zookeeper.properties然后再启动 Kafka$KAFKA_HOME/bin/kafka-server-start.sh -daemon $KAFKA_HOME/config/server.properties-daemon参数让它在后台运行。启动后别急着操作等 5-10 秒然后用jps看一下有没有Kafka进程。如果进程一闪而过就没了去$KAFKA_HOME/logs/server.log里看错误日志大概率是 ZooKeeper 连接问题或者端口占用。注意如果你用的是 Docker 环境Kafka 的listeners配置和宿主机访问之间需要额外处理。容器内监听 9092宿主机要映射端口同时advertised.listeners要设成宿主机能访问的地址否则宿主机上的客户端连不上。这是 Docker 里跑 Kafka 最常见的坑。3.3 用命令行验证消息收发完整链路走一遍配置完不验证等于没配。Kafka 自带了一套命令行工具可以完整地模拟生产者和消费者。先创建一个测试 topickafka-topics.sh --create --topic test-topic \ --bootstrap-server localhost:9092 \ --partitions 3 --replication-factor 1注意这里用的是--bootstrap-server而不是老版本的--zookeeper。Kafka 2.2 之后推荐用 bootstrap-server 方式直接连 Broker 而不是通过 ZooKeeper 转发。创建完用--describe看一下kafka-topics.sh --describe --topic test-topic --bootstrap-server localhost:9092你应该能看到 3 个分区每个分区的 Leader 都是 0因为只有一个 Broker副本数都是 1。然后开两个终端窗口一个做生产者kafka-console-producer.sh --topic test-topic --bootstrap-server localhost:9092另一个做消费者kafka-console-consumer.sh --topic test-topic --bootstrap-server localhost:9092 --from-beginning在生产者窗口里敲几行文字回车消费者窗口应该能实时看到。如果消费者收不到检查两点一是--from-beginning有没有加不加的话只消费启动之后的新消息二是 topic 名称有没有打错。这个验证过程看起来简单但它确认了三件事Broker 正常启动、ZooKeeper 协调正常、生产消费链路通畅。这三件事任何一件出问题后面接 Spark Streaming 或者 Flink 都会失败。3.4 伪分布式下 Kafka 的性能调优小技巧虽然是学习环境但了解几个关键参数对理解 Kafka 的运行机制很有帮助。日志分段策略Kafka 的消息是追加到日志段文件里的默认每个段 1GB。在伪分布式环境里磁盘空间有限可以调小到 256MB通过log.segment.bytes设置。段文件小了清理和压缩的粒度更细。消息保留时间默认是 168 小时7 天。学习环境里数据量小可以改成 24 小时通过log.retention.hours24设置避免磁盘被测试数据塞满。分区数的影响前面设了 3 个分区你可以做个实验——用生产者发 100 条消息然后看每个分区的偏移量。用kafka-run-class.sh kafka.tools.GetOffsetShell --topic test-topic --bootstrap-server localhost:9092能看到每个分区的当前最大偏移量。如果三个分区的偏移量差不多说明消息分布均匀如果差很多说明你的消息 key 设置有问题没有 key 时默认轮询有 key 时按 key 哈希。这些参数在真正的分布式集群里调优逻辑更复杂但在伪分布式里先把概念建立起来后面扩展就顺了。4. Sqoop 的部署与关系型数据库到 HDFS 的搬运实战4.1 Sqoop 安装中的 jar 包补全策略Sqoop 的安装包解压之后lib目录下的 jar 并不完整尤其是连接各种数据库的驱动 jar 需要你自己放。这是新手最容易卡住的地方——Sqoop 命令能跑但一连接 MySQL 就报ClassNotFoundException: com.mysql.jdbc.Driver。完整的操作步骤是这样的tar -zxvf sqoop-1.4.7.bin__hadoop-2.6.0.tar.gz -C /opt/ mv /opt/sqoop-1.4.7.bin__hadoop-2.6.0 /opt/sqoop然后配置环境变量除了SQOOP_HOME和 PATH还要确保HADOOP_HOME已经配好因为 Sqoop 启动时会去读 Hadoop 的配置export SQOOP_HOME/opt/sqoop export PATH$PATH:$SQOOP_HOME/bin接下来是关键的驱动 jar 处理。去 MySQL 官网下载mysql-connector-java的 jar 包5.1.x 或 8.0.x 都行看你的 MySQL 版本放到$SQOOP_HOME/lib下。如果你用的是 Hadoop 3.x还需要处理前面提到的 Guava 冲突# 查看 Hadoop 用的 Guava 版本 ls $HADOOP_HOME/share/hadoop/common/lib/ | grep guava # 假设是 guava-27.0-jre.jar把它复制到 Sqoop lib 下替换旧版 cp $HADOOP_HOME/share/hadoop/common/lib/guava-27.0-jre.jar $SQOOP_HOME/lib/ # 删掉 Sqoop 自带的旧版 guava rm $SQOOP_HOME/lib/guava-*.jar # 注意别把刚拷进去的删了先确认文件名这里有个操作细节rm的时候一定要先ls看清楚文件名别把刚复制进去的新版误删了。我建议的做法是先把旧版重命名成.bak确认新版能用之后再删。验证 Sqoop 是否可用sqoop version如果输出了版本号并且没有报错说明基础环境 OK。然后测试连接 MySQLsqoop list-databases \ --connect jdbc:mysql://localhost:3306/ \ --username root \ --password yourpassword能列出数据库列表说明驱动和连接都没问题。4.2 从 MySQL 导入数据到 HDFS参数背后的逻辑Sqoop 导入的核心命令是sqoop import但参数很多我挑最关键的几个讲清楚它们的作用和选择理由。假设 MySQL 里有个user_db数据库里面有个users表我们要把它导入到 HDFS 的/sqoop/users目录sqoop import \ --connect jdbc:mysql://localhost:3306/user_db \ --username root \ --password yourpassword \ --table users \ --target-dir /sqoop/users \ --num-mappers 4 \ --split-by id \ --fields-terminated-by \t逐个解释--num-mappers 4指定用 4 个 Map 任务并行导入。Sqoop 会根据--split-by指定的列把数据切成 4 份每个 Map 处理一份。这个值不是越大越好一般设成和集群可用容器数匹配。伪分布式里 NodeManager 可能只配了 2-4 个容器设太大反而排队。--split-by id切分列必须是数值型或者能均匀分布的列。如果选了一个分布不均的列比如性别会导致数据倾斜某个 Map 处理 90% 的数据其他 Map 很快就跑完了整体时间反而更长。--fields-terminated-by \t指定导出到 HDFS 后的字段分隔符。默认是逗号但如果你的数据里本身有逗号就会解析错乱。用制表符更安全。导入完成后去 HDFS 上看一下hdfs dfs -ls /sqoop/users hdfs dfs -cat /sqoop/users/part-m-00000 | head -5你应该能看到 4 个 part 文件对应 4 个 Map每个文件里是制表符分隔的数据。提示如果导入时报 Cannot run program sqoop 或者 MapReduce 作业卡住检查HADOOP_MAPRED_HOME环境变量有没有设。Sqoop 需要这个变量来找到 MapReduce 的 jar 包。设成$HADOOP_HOME即可。4.3 导出方向HDFS 数据回流到 MySQL 的注意事项Sqoop 的导出sqoop export是把 HDFS 上的数据写回关系型数据库。这个方向在实际业务里用得也很多比如把 Hive 分析后的结果表同步回 MySQL 供报表系统查询。sqoop export \ --connect jdbc:mysql://localhost:3306/user_db \ --username root \ --password yourpassword \ --table users_result \ --export-dir /sqoop/users_result \ --input-fields-terminated-by \t \ --num-mappers 2导出比导入更容易出问题主要有三个坑第一目标表必须提前建好。Sqoop 不会自动建表如果users_result表不存在直接报错。建表时字段类型要和 HDFS 数据匹配比如 HDFS 里是字符串的日期MySQL 里对应VARCHAR而不是DATE否则会插入失败。第二分隔符必须和 HDFS 文件一致。导入时用了\t导出时就要用--input-fields-terminated-by \t不匹配的话整行数据会被当成一个字段插入报错。第三主键冲突处理。如果目标表有主键而 HDFS 数据里有重复主键导出会失败。可以用--update-key指定更新列或者--update-mode allowinsert让冲突时执行更新而非插入。4.4 把 Sqoop 任务串进调度流程的思路单次执行 Sqoop 命令只是开始实际项目里你需要把它串进调度流程。最简单的做法是写一个 shell 脚本把多个 Sqoop 命令按顺序排列加上错误检查和日志记录#!/bin/bash LOG_FILE/var/log/sqoop_import_$(date %Y%m%d).log echo 开始导入 users 表: $(date) $LOG_FILE sqoop import \ --connect jdbc:mysql://localhost:3306/user_db \ --username root \ --password yourpassword \ --table users \ --target-dir /sqoop/users/$(date %Y%m%d) \ --num-mappers 4 \ --split-by id $LOG_FILE 21 if [ $? -eq 0 ]; then echo 导入成功: $(date) $LOG_FILE else echo 导入失败请检查日志: $(date) $LOG_FILE exit 1 fi这个脚本可以挂到 crontab 里定时执行也可以用 Airflow、Oozie 这类调度工具管理。关键点是每次导入到带日期的目录这样历史数据不会互相覆盖也方便回溯。5. 组件之间的联动让 Kafka 和 Sqoop 融入现有数据链路5.1 一个完整的数据流转场景设计单独会用一个组件不算本事能把它们串起来解决实际问题才是目的。我设计一个在伪分布式环境里可以完整跑通的场景帮你理解各组件怎么配合。场景是这样的假设有一个模拟的业务系统用户注册信息存在 MySQL 里用户行为日志实时产生。我们要做的是用 Sqoop 每天凌晨把 MySQL 里的用户全量数据导入 HDFS再加载到 Hive 分区表。用户行为日志通过一个生产者程序实时写入 Kafka。用一个消费者程序从 Kafka 读数据落地到 HDFS 的另一个目录。在 Hive 里把两份数据关联分析。这个场景覆盖了离线批量Sqoop和实时流式Kafka两条链路是典型的大数据 Lambda 架构的简化版。5.2 Sqoop 导入的数据如何对接 HiveSqoop 可以直接导入到 Hive用--hive-import参数sqoop import \ --connect jdbc:mysql://localhost:3306/user_db \ --username root \ --password yourpassword \ --table users \ --hive-import \ --hive-table user_db.users \ --hive-partition-key dt \ --hive-partition-value 20240101 \ --num-mappers 4 \ --split-by id--hive-import会自动在 Hive 里建表如果不存在然后把数据加载进去。--hive-partition-key和--hive-partition-value指定分区这对按天导入的场景非常实用。但这里有个坑Sqoop 自动建的表字段类型可能不是你想要的。比如 MySQL 的VARCHAR(255)会被映射成 Hive 的STRING这通常没问题但DATETIME可能被映射成STRING而不是TIMESTAMP后续做时间计算时要手动转换。我的建议是先手动在 Hive 里建好表然后用--hive-table指定已存在的表加上--hive-overwrite覆盖数据这样字段类型完全可控。5.3 Kafka 消费者落地 HDFS 的简易实现从 Kafka 消费数据写到 HDFS可以用 Spark Streaming 或者 Flink但在伪分布式环境里跑这些框架比较重。一个轻量级的替代方案是写一个 Python 脚本用kafka-python消费然后定期把数据追加到 HDFS 文件。核心逻辑是消费者持续拉取消息攒够一定条数或者超过一定时间就写一次 HDFS。写 HDFS 可以用hdfs命令行也可以用pyarrow的 HDFS 接口。用命令行最简单from kafka import KafkaConsumer import subprocess import json import time consumer KafkaConsumer( user-behavior, bootstrap_servers[localhost:9092], auto_offset_resetearliest, group_idhdfs-writer ) buffer [] last_flush time.time() for message in consumer: buffer.append(message.value.decode(utf-8)) # 攒够 100 条或者超过 30 秒就写一次 if len(buffer) 100 or (time.time() - last_flush) 30: local_file /tmp/kafka_buffer.txt with open(local_file, w) as f: f.write(\n.join(buffer)) # 追加到 HDFS subprocess.run([ hdfs, dfs, -appendToFile, local_file, /kafka/user-behavior/behavior.log ]) buffer [] last_flush time.time()这个脚本虽然简陋但把 Kafka 到 HDFS 的链路跑通了。实际生产里你会用 Flink 的StreamingFileSink或者 Spark 的writeStream但原理是一样的消费、缓冲、批量写入。注意hdfs dfs -appendToFile在 HDFS 上频繁调用会有性能问题因为每次都要和 NameNode 通信。生产环境要用专门的流式写入连接器。学习环境里数据量小这样写没问题。5.4 验证整条链路是否通畅的检查清单链路搭好之后按这个清单逐项验证检查项验证命令预期结果Sqoop 能连 MySQLsqoop list-databases --connect ...列出数据库名Sqoop 导入 HDFShdfs dfs -ls /sqoop/users看到 part 文件Hive 能查导入数据hive -e select count(*) from user_db.users返回行数Kafka 生产消费正常前面 3.3 节的命令行测试消费者收到消息Kafka 数据落地 HDFShdfs dfs -cat /kafka/user-behavior/behavior.log看到消息内容Hive 关联查询hive -e select ... join ...返回关联结果每一项都通过之后这套伪分布式集群就算真正具备了完整的数据处理能力。任何一项失败回到对应章节排查。6. 踩坑复盘那些让我折腾半天的典型问题6.1 Sqoop 导入时卡在 map 0% reduce 0%这个现象我遇到过好几次表现是 Sqoop 命令执行后MapReduce 进度一直卡在 0%既不报错也不前进。原因通常有三个YARN 资源不足。Sqoop 提交的 MapReduce 作业在 YARN 上申请不到容器一直处于 ACCEPTED 状态。去 ResourceManager Web UI 看作业状态如果是 ACCEPTED 而不是 RUNNING就是资源问题。解决办法是调大yarn.nodemanager.resource.memory-mb和yarn.scheduler.maximum-allocation-mb或者减少--num-mappers。数据库连接超时。Sqoop 在切分数据之前要先连数据库查一次元信息如果数据库地址不通或者防火墙拦截会卡在连接阶段。用telnet mysql_host 3306测试连通性。split-by 列有 NULL 值。Sqoop 切分数据时如果切分列有 NULL边界计算会出问题导致某些 Map 任务拿不到数据范围而卡住。解决办法是选一个没有 NULL 的列做 split-by或者在 SQL 里过滤掉 NULL。6.2 Kafka 启动后立即退出且日志无明显错误Kafka 启动脚本用-daemon后进程消失日志里只有 INFO 级别的信息没有 ERROR。这种情况最让人头疼因为没有错误本身就是最大的错误。排查思路是这样的先看server.log的最后 20 行找有没有 shutdown 或者 fatal 关键字。如果没有去看$KAFKA_HOME/logs/下有没有kafkaServer.out文件标准输出和错误输出可能被重定向到了那里。最常见的原因是ZooKeeper 连接失败Kafka 启动时要连 ZooKeeper 注册自己连不上就退出。用zkCli.sh -server localhost:2181测试 ZooKeeper 是否可达。log.dirs 目录权限问题Kafka 进程的用户对log.dirs没有写权限启动时创建日志目录失败。chmod或chown解决。端口被占用9092 被其他进程占了Kafka 绑定失败。netstat -tlnp | grep 9092查一下。6.3 Docker 环境下宿主机访问 Kafka 的地址配置如果你是在 Docker 容器里跑 Kafka宿主机上的客户端要连进来advertised.listeners必须配成宿主机能解析的地址。很多人只配了listenersPLAINTEXT://0.0.0.0:9092容器内能连宿主机连就报 Connection refused。正确的配置是listenersPLAINTEXT://0.0.0.0:9092 advertised.listenersPLAINTEXT://宿主机IP:9092同时 Docker 启动时要-p 9092:9092映射端口。宿主机 IP 不能写localhost因为对宿主机来说localhost是宿主机自己不是容器。写宿主机的实际局域网 IP 或者host.docker.internalMac 和 Windows 的 Docker Desktop 支持。6.4 Sqoop 导出时中文乱码的处理从 HDFS 导出中文数据到 MySQL如果两边字符集不一致会出现乱码。HDFS 文件默认是 UTF-8MySQL 表如果建的时候没指定字符集可能是 latin1。解决办法是在 JDBC 连接串里显式指定字符集--connect jdbc:mysql://localhost:3306/user_db?useUnicodetruecharacterEncodingutf8同时确认 MySQL 表的字符集是utf8mb4。建表时用CREATE TABLE ... DEFAULT CHARSETutf8mb4。这两个地方都对了中文就不会乱码。7. 后续可以继续折腾的方向这套环境跑通之后其实还有很多可以深挖的地方。比如把 Kafka 的消息用 Spark Streaming 消费做实时词频统计或者窗口聚合或者用 Sqoop 的--incremental append模式做增量导入只同步 MySQL 里新增的数据而不是全量再或者把 Hive 的分析结果通过 Sqoop 导出回 MySQL形成一个闭环。我个人觉得最有价值的扩展方向是把 Sqoop 的定时导入和 Kafka 的实时消费结合起来做 Lambda 架构的完整实验。离线层用 Sqoop Hive 保证数据的完整性和准确性实时层用 Kafka Spark Streaming 保证低延迟最后在应用层做合并查询。这个架构在真实业务里非常常见在伪分布式环境里完整跑一遍对理解大数据系统的运作方式帮助极大。另外如果你对性能调优感兴趣可以试着调整 Kafka 的分区数和 Sqoop 的并行度观察不同配置下任务的执行时间变化。这种改一个参数、跑一次任务、记录一次结果的实验方法比看多少文档都管用。