物联网数据这几年我经手了不少，从车联网终端上报的轨迹点，到厂房里各种PLC传感器采集的温度、振动、能耗数据，说白了就是一个字：多。设备一多、频率一高，一天少说几千万条记录，传统的关系型数据库根本扛不住——不是存不下，是查询和统计能把人卡到怀疑人生。后来我把整套处理链路迁到 Hadoop 生态上，用 HDFS 做统一存储、Spark 做批量清洗和聚合、Hive 做即席查询，才算把这块硬骨头啃了下来。

这篇东西不是教科书式科普，是我从零搭建、踩坑、上线、调优的一次完整复盘。不管你是刚接触大数据的学生，还是想给团队的物联网数据找个靠谱处理方案，都可以照着这里的思路走一遍。

1. 物联网数据与 Hadoop 的适配逻辑

1.1 物联网数据到底“脏”在哪

很多人一提到物联网，第一反应是传感器很智能。但实际拿到手里的物联网原始数据，往往是一堆“半成品”：设备离线产生的补传数据时间戳错乱，信号干扰导致数值突然跳变，不同厂商设备的字段命名千奇百怪，有的上报 JSON、有的只传 CSV，还有的直接把二进制协议解析后的字符串扔过来。更麻烦的是，同一批设备里还混着大量重复上报和无效点位。

这些数据天然有三个特点：体量大、乱序多、价值密度低。体量大意味着单机存储和计算都吃不消；乱序多意味着清洗时必须有全局排序和去重逻辑；价值密度低意味着我们往往要先把原始明细存下来，之后反复跑不同口径的统计，不能只留一个压缩过的汇总表。Hadoop 的 HDFS 恰好能低成本地把原始数据全量留存，MapReduce 和 Spark 这类批计算引擎又适合做全量扫描和清洗，所以从底层架构上讲，物联网数据放在 Hadoop 生态里是成立的。

1.2 Hadoop 三大组件各管哪一段

Hadoop 给人的印象是一个“大箱子”，但实际拆开看，核心就三块：HDFS、YARN、计算引擎。

• HDFS 管存储：把大文件切块，分布式地放在多个节点上，并且默认复制三份，防止某台机器挂掉丢数据。物联网场景下，一天的原始日志可以到几十 GB，直接丢 HDFS 里，不需要提前设计什么分库分表。
• YARN 管资源：它像一个大管家，谁要跑计算任务，就给它分配多少 CPU 和内存。多个计算引擎可以同时跑在一个集群上，互不打架。
• 计算引擎管逻辑：老一代是 MapReduce，现在生产环境基本都用 Spark、Hive 或 Flink 来处理。MapReduce 虽然执行慢，但胜在稳定，适合跑夜间批量任务；Spark 执行效率高，适合做多次迭代的清洗和聚合。

这套架构解决的核心问题，是把“一台机器硬扛”换成“一群机器分工协作”。你不需要关心某条数据存在哪个节点的哪块磁盘上，只需要把任务提交上去，框架自己会找数据所在的节点去算，也就是常说的 data locality——移动计算而不是移动数据。数据量越大的时候，这个优势越明显。

2. 从零搭建可用的 Hadoop 环境

2.1 单机伪分布式与真实集群的选择

第一次上手 Hadoop，没必要一开始就上五台物理机，先用“伪分布式”把流程跑通是性价比最高的方式。所谓伪分布式，就是在一台 Linux 机器上同时启动 NameNode、DataNode、ResourceManager、NodeManager 这几个进程，模拟出一个“一节点集群”。

我一般建议新手用 Ubuntu 虚拟机加三到四个节点来做。伪分布式适合验证代码和熟悉命令，但如果你要测数据分片、节点宕机后的容错，或者 YARN 的资源调度，最少得搭一个三节点集群：一个 NameNode 节点和两个 DataNode 节点。这一步千万别图省事只搭伪分布式，因为真实环境里的很多坑，比如数据块副本不足导致文件变成 corrupt 状态、节点之间 hostname 解析不通，只有在多节点下才会暴露。

搭建时注意几个关键点：

• 所有节点使用统一的 Linux 用户，并且配置 SSH 免密登录。否则每次启动集群都要输密码，启动脚本根本没法用。
• /etc/hosts里必须把集群所有节点的 IP 和 hostname 对应关系写全。很多人第一次搭集群失败，就是因为主机名解析不对，节点之间互相找不到。
• 配置 Java 环境变量时，Hadoop 3.x 需要 Java 8 以上，我建议直接用 Java 8，兼容性最稳。

2.2 核心配置项与内存参数

配置文件主要改三个：core-site.xml、hdfs-site.xml、yarn-site.xml。新手最容易犯的错是照搬网上的配置，不看自己的内存大小，导致 NodeManager 申请内存超过物理机内存，启动后直接被系统杀掉。

我个人的经验做法是：物理机或虚拟机内存 8 GB 的话，给 NameNode 分配 1 GB，DataNode 1 GB，ResourceManager 1 GB，NodeManager 2 GB，剩下留给操作系统和后续跑的 Spark 任务。yarn-site.xml里把yarn.nodemanager.resource.memory-mb设为 2048，yarn.nodemanager.resource.cpu-vcores设为 2，这样至少能稳定跑起来两个容器。

伪分布式的核心配置如下：

<!-- core-site.xml --> <configuration> <property> <name>fs.defaultFS</name> <value>hdfs://localhost:9000</value> </property> </configuration> <!-- hdfs-site.xml --> <configuration> <property> <name>dfs.replication</name> <value>1</value> </property> </configuration> <!-- yarn-site.xml --> <configuration> <property> <name>yarn.nodemanager.aux-services</name> <value>mapreduce_shuffle</value> </property> <property> <name>yarn.nodemanager.resource.memory-mb</name> <value>2048</value> </property> </configuration>

伪分布式里dfs.replication必须设 1，因为只有一台 DataNode，默认的 3 份副本会一直处于 under-replicated 状态，看着很烦。多节点集群再改回 3。

启动顺序别搞错：先start-dfs.sh，再start-yarn.sh。然后输入jps，能看到 NameNode、DataNode、ResourceManager、NodeManager 这 4 个进程就说明基础环境通了。

3. 把物联网数据送进 HDFS

3.1 数据落地路线：Flume 与脚本导入

数据从设备端到 HDFS，常见路线有三条：

• 设备端直接推送到 Kafka，再由 Flume 或自研消费程序写入 HDFS。这条链路最接近生产环境，适合需要缓冲削峰的场景。
• 设备端产生的文件按小时落到边缘网关，网关上的脚本定时调用hdfs dfs -put上传。
• 如果已经有采集服务在写 MySQL，再用 Sqoop 每天把增量数据同步到 HDFS。这种做法适合业务系统已经跑了一段时间、想拿历史数据做分析的团队。

我在一个交通信息分析的小项目里用的是第二种，网关每 5 分钟生成一个包含车辆 GPS 点位和速度的 JSON 文件，然后由一个 shell 定时任务把文件压缩成 gzip 后传到 HDFS 指定目录。之所以先压缩再上传，是因为 GPS 轨迹点这类文本数据压缩率极高，10 GB 原始数据 gzip 后往往只剩 1.5 GB 左右，能大幅节省网络带宽和存储成本。

压缩格式的选择上，我建议优先用 gzip。虽然它不支持切分，但物联网设备上报处理逻辑简单，每个业务目录下本身就按小时分好了文件，单文件多压缩几倍后可能只有十几 MB，切分需求不明显。只有当单个文件超过 HDFS 块大小（默认 128 MB）时，你才需要考虑用支持切分的 LZO 或 snappy。

3.2 文件格式与压缩比

物联网数据推荐用列式存储还是行式存储？我踩过坑，这里给个明确建议：如果走 Spark SQL 或 Hive，最终表用 Parquet + snappy 压缩；但如果只是把原始数据先原样留存，就保留 JSON 或者 CSV 的 gzip。

Parquet 这种列式格式在查询时只读需要的列，性能比 CSV 高一个量级。但列式存储的前提是你已经完成了数据清洗，字段统一、类型统一。原始物联网数据还没清洗前，字段可能有缺失，类型也可能不对，强行转 Parquet 反而会引入很多解析错误。所以我一贯的原则是：原始层用什么格式无所谓，干净层必须要用 Parquet。

把 JSON 转成 Parquet，推荐直接用 Spark 读进来再写出去，几行代码就搞定：

val rawDF = spark.read.json("/data/iot/raw/2024/06/01") val cleanedDF = rawDF.select( $"device_id", $"event_time".cast("timestamp"), $"longitude".cast("double"), $"latitude".cast("double"), $"speed".cast("double") ) cleanedDF.write.mode("overwrite") .partitionBy("dt") .format("parquet") .save("/data/iot/clean/2024/06/01")

这段代码的核心逻辑是先加载当天的原始 JSON，只挑出后续分析要用到的字段并转换成强类型，再按日期分区写入 Parquet。一旦走到这一步，后面的统计任务就都在这份干净数据上跑了。

4. 用 MapReduce 模型做第一版清洗与聚合

4.1 按设备 ID 聚合的 MapReduce

接触一个新框架，第一件事不是背 API，而是跑通一个能解决实际问题的例子。网上那些 wordcount 例子用途不大，我建议直接写一个“统计每台设备当天上报了多少条点位”的作业，流程和 wordcount 一模一样，但更贴近物联网场景。

任务很具体：从原始 JSON 里提取device_id和event_time，按小时统计每台设备的点位数量，找出上报异常偏少的设备。用 MapReduce 实现时，Mapper 负责解析 JSON，把键设为device_id、值设为 1，Reducer 负责求和。

核心代码长这样：

public class DeviceCountMapper extends Mapper<Object, Text, Text, IntWritable> { private final static IntWritable one = new IntWritable(1); private Text deviceId = new Text(); public void map(Object key, Text value, Context context) throws IOException, InterruptedException { String line = value.toString(); // 每行是一条 JSON，解析 field 可以引入 fastjson 或自写简单抽取 String device = parseDeviceId(line); deviceId.set(device); context.write(deviceId, one); } } public class DeviceCountReducer extends Reducer<Text, IntWritable, Text, IntWritable> { private IntWritable result = new IntWritable(); public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException { int sum = 0; for (IntWritable val : values) { sum += val.get(); } result.set(sum); context.write(key, result); } }

这个作业跑完后，你会得到一个明细输出：每台设备对应的上报总数。但要注意，如果某台设备一整天都没上报，它根本不会出现在输出里，因为 MapReduce 只处理存在的数据。要找出“完全没上报的设备”，还需要拿设备元数据表做一次外连接，这通常放到 Hive 里处理更合适。

4.2 任务提交与日志排查

作业写好后，打包成 jar，用hadoop jar命令提交到 YARN 上运行：

hadoop jar iot-process.jar com.example.DeviceCount \ /data/iot/raw/2024/06/01 \ /data/iot/result/device_count/2024/06/01

我每次跑任务都会先跟踪一下进度，用yarn application -status或直接打开 ResourceManager 的 Web 页面看 container 日志。新手常见的问题是代码一跑就内存溢出，但不知道去哪儿看。MapReduce 里 Mapper 的内存溢出通常表现为 task 反复重试，看日志时重点盯着mapreduce.map.memory.mb和mapreduce.reduce.memory.mb这两个参数。默认值在 3.x 里是 1 GB 左右，如果你每条 JSON 里嵌了一个很大的字段，比如把设备上报的整包数据都放到 value 里，1 GB 很容易被打满。

4.3 合并小文件

物联网数据一落地就容易产生大量小文件：网关每 5 分钟一个文件，一天就是几百个，一个月就是上万个。小文件多了，NameNode 内存会崩溃，因为每个文件都要在内存里维护元数据；跑任务时每个文件还要对应一个 task，启动 task 的开销比计算本身还大。

我的处理手段是定期用 Spark 做一次“小文件重分区”：

spark.read.parquet("/data/iot/clean") .repartition(24) // 按一天 24 小时控制输出文件数 .write.mode("overwrite") .partitionBy("dt") .option("compression", "snappy") .format("parquet") .save("/data/iot/clean_merged")

这里repartition(24)的意义是把当天数据重新打散为固定数量的 24 个大文件，而不是按分区目录原样保留几百个小文件。但这句话有个前提——你要清楚自己集群的块大小。如果 24 个文件每个才 5 MB，那分得还是太细。正确的做法是估算当天总数据量，比如 2 GB，想让每个文件在 128 MB 左右，那就repartition(16)左右。宁可多试几次把文件数调小，也不要每个文件几 MB，那样跑数仓任务会慢得让人抓狂。

5. 升级为 Spark SQL + Hive 的工业级处理

5.1 为什么最终迁移到 Spark/Hive

MapReduce 作业写起来啰嗦，而且每次迭代都要把中间结果写回磁盘，跑多步骤清洗链路时慢得恼人。后来我把核心链路迁移到了 Spark SQL + Hive 上：用 Hive 管理表结构，用 Spark SQL 做查询和计算。

迁移带来的收益很明显：

• Spark 基于内存计算，多阶段的 ETL 任务比 MapReduce 快 3 到 10 倍。
• Spark SQL 内置大量函数，处理 JSON、时间窗口、开窗去重都比手写 MapReduce 方便得多。
• Hive 的元数据服务让所有表结构统一，Spark 可以直接spark.sql("select ... from xxx where ...")，不需要自己写文件路径解析逻辑。

Hive 的另一个隐藏价值是它把存储路径变成了类似关系数据库的表结构。比如数据放在 HDFS 的/warehouse/iot.db/device_report目录下，你在 Hive 里建一张外层表，指定 location 指向这个目录，执行 SQL 时就不用关心底层文件是怎么组织的了。

5.2 建表、分区与 ZooKeeper 在分布式协调中的角色

用 Hive 管理物联网数据的核心是分区设计。我常用的分区字段是dt，按天分区。这样做的好处是查询时可以直接裁剪掉无关分区，只扫需要的日期，效率提升非常明显。

建表示例：

CREATE EXTERNAL TABLE iot.device_report ( device_id STRING, event_time TIMESTAMP, longitude DOUBLE, latitude DOUBLE, speed DOUBLE, temperature DOUBLE ) PARTITIONED BY (dt STRING) STORED AS PARQUET LOCATION '/data/iot/warehouse/device_report';

这是外表，数据文件已经存在 HDFS 上，Hive 只负责挂元数据。跑任务前先执行一条MSCK REPAIR TABLE device_report;让 Hive 自动识别 HDFS 上新出现的分区目录，否则查不到数据。这个命令我每次都会忘一次，然后被同事吐槽。

集群规模稍微上来以后，Hive 和 HBase 这类服务经常同时跑在一个集群上。多节点分布式环境下，各节点之间的状态同步需要协调者。这里就是 ZooKeeper 登场的地方：HDFS 的 NameNode 高可用依赖 ZooKeeper 做主备切换，YARN 的 ResourceManager 高可用同样依赖它。实际生产里，我见过把 ZooKeeper 单独用三台机器搭成一个 ensemble 的，也见过和 Hadoop 节点混部，看机器资源情况而定。最关键的配置是zoo.cfg里的server.1、server.2、server.3三行地址要写对，并且每个节点都要配一个数据目录，不能共享一个目录，否则启动后互相抢锁，状态一片混乱。

5.3 交通信息统计案例

拿我之前做的交通信息分析系统举例。车辆的 GPS 轨迹数据进 HDFS 以后，最早是直接查明细，慢得要命。后来改成 Hive 里维护一张“车辆时段聚合表”，每天夜间用 Spark 作业跑一次，统计每辆车在每一小时内的平均速度、最大速度、轨迹点数量和经过的路段标识。

核心 SQL 大概是：

INSERT OVERWRITE TABLE iot.device_hourly_summary PARTITION (dt) SELECT device_id, dt, hour(event_time) AS hour, count(*) AS point_count, round(avg(speed), 2) AS avg_speed, max(speed) AS max_speed FROM iot.device_report WHERE dt = '2024-06-01' GROUP BY device_id, dt, hour(event_time);

跑完这条 SQL，每天的数据量从几千万条明细压缩成几十万条汇总，后续做可视化、出报表，直接查这张聚合表就行，几十毫秒内能出结果。这个思路是物联网数据处理里最核心的一招：明细层永远保留原始数据，汇总层只留指标。分析需求变的时候，从明细层重新跑一套聚合就行，不需要回设备端重新采集。

6. 典型故障排查与常见问题速查

6.1 经常出现的错误

跑 Hadoop 和 Spark 的几个常见问题，基本每个人都会碰到：

• Could not find or load main class或者提交作业后一直显示RUNNING但没有任何输出。多半是 jar 包依赖没打全，或者 worker 节点上的 Spark 环境变量没配一致。我惯用spark-submit --master yarn --deploy-mode cluster时把依赖包打进 fat jar，本地模式没问题、集群模式就报 ClassNotFound，基本都是这个原因。
• DataNode 启动后过几分钟进程消失。先看日志，常见是磁盘空间不足或者dfs.data.dir指向了不存在的目录。
• MapReduce 任务 100% map 完成，reduce 一直停在 33.33%。不要慌，大概率是 reducer 正在拉取 map 输出，网络或磁盘速度慢而已；但如果一直卡着不动，去查 reducers 节点是不是内存碎片太多。
• Hive 查询返回结果为空，明明 HDFS 上有文件。先跑dfs -ls看路径有没有权限问题，再MSCK REPAIR刷新分区，最后检查是不是 iot 表存的是 rename 前的旧路径。

6.2 面试和架构设计里常被问到的点

不少读者是学生，面试大数据岗位时经常被问“说说你做过的一个 Hadoop 项目”。我建议不要只背理论，把一个物联网数据处理项目讲透。面试官常追问的点其实就两个方向：一是你如何处理数据倾斜，二是你如何保证数据不丢不重。

数据倾斜在物联网数据分析里非常常见。比如统计设备活跃度时，某几个头部设备的点位特别多，reduce 阶段就一个 task 卡到天荒地老。解决办法，我之前用的是加盐打散：先把 key 后面拼上一个随机数，把数据分成 10 份分别聚合，第二步再按真实 key 聚合一次。代价是跑两轮任务，但稳定性好了很多。

数据不丢不重里，最值得注意的坑是“重跑作业时覆盖写”。写入 HDFS 时用overwrite模式一定要谨慎，如果下游已经在读这份数据，你覆盖的同时下游可能读到半份文件。我的习惯是每次跑批都先把结果写到临时目录，成功后再把临时目录原子地 rename 成正式目录，能避免很多诡异问题。

收尾的经验之谈

要是把整个流程压缩成一句话，那就是：物联网数据处理的关键不在技术栈有多高级，而在于把“原始数据沉淀”和“指标计算”分层做好。Hadoop 这层架构，最大的价值是给你提供了一个稳定、能扩展的底座，让你不用每天担心“机器磁盘是不是又满了”“数据要不要删一部分腾空间”。我做的项目里，从最初的单机脚本，到后来三节点集群加 Hive 加 ZooKeeper 协调，整个演进过程花了大概两个月，最耗时间的部分反而不是写代码，而是调内存参数和排查各种分布式环境下的怪毛病。

最后分享一个小技巧：拿到一批物联网数据，先别急着写清洗逻辑，花半小时把数据可能存在的异常列个清单——时间乱序、设备 ID 重复、空字段、超范围数值，写清楚每种异常对应什么处理规则，然后才开始建表和写 ETL。这个清单看起来不起眼，但它能让你少走非常多的弯路，也能让你在跟同事讨论需求时更有底气。