Ubuntu Hadoop实时数据处理怎么做

Ubuntu环境下实现Hadoop实时数据处理的技术方案
传统Hadoop MapReduce以批处理为核心，无法满足实时数据处理需求。需结合流处理框架（如Flink、Spark Streaming）与实时数据管道（如Kafka），构建“采集-传输-处理-存储”的完整实时流水线。以下是具体实施步骤：

1. 环境准备：安装基础组件

安装Java环境：Hadoop及流处理框架依赖Java，通过以下命令安装OpenJDK 11（推荐）：
```
sudo apt update
sudo apt install openjdk-11-jdk
```
验证安装：java -version（需显示Java版本信息）。
安装Hadoop：从官网下载稳定版（如3.3.1），解压至指定目录（如/usr/local/hadoop-3.3.1），配置核心文件：
- core-site.xml：设置HDFS默认文件系统地址（fs.defaultFS=hdfs://localhost:9000）；
- hdfs-site.xml：配置NameNode数据目录（dfs.namenode.name.dir=/opt/hadoopdata/namenode）与DataNode数据目录（dfs.datanode.data.dir=/opt/hadoopdata/datanode），并设置副本数（dfs.replication=1，单机环境）。
  格式化NameNode：hadoop namenode -format，启动HDFS：start-dfs.sh（通过jps验证NameNode、DataNode进程是否运行）。

2. 构建实时数据管道：Kafka数据采集

Kafka作为分布式消息队列，负责实时数据的接收、存储与传输，是连接数据源与流处理框架的关键组件。

安装Kafka：下载最新版（如2.8.1），解压后启动ZooKeeper（Kafka依赖）与服务：

bin/zookeeper-server-start.sh config/zookeeper.properties &
bin/kafka-server-start.sh config/server.properties &

创建Topic：用于分类存储数据流（如test_topic）：

bin/kafka-topics.sh --create --topic test_topic --bootstrap-server localhost:9092 --partitions 1 --replication-factor 1

验证数据传输：通过控制台生产者发送消息（bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test_topic），消费者接收消息（bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test_topic --from-beginning）。

3. 选择实时处理框架：Flink/Spark Streaming

方案A：Apache Flink（推荐，低延迟+精确一次处理）

Flink是原生流处理框架，支持事件时间、状态管理与容错，适合实时分析、实时ETL等场景。

安装Flink：下载稳定版（如1.13.2），解压后配置环境变量（~/.bashrc添加export FLINK_HOME=/usr/local/flink-1.13.2、export PATH=$PATH:$FLINK_HOME/bin），启动集群：
```
bin/start-cluster.sh
```
验证：访问http://localhost:8081查看Flink Web UI。

编写Flink实时处理程序（Python示例）：使用Flink的DataStream API读取Kafka数据，进行单词计数：

from pyflink.datastream import StreamExecutionEnvironment
from pyflink.datastream.connectors.kafka import FlinkKafkaConsumer
from pyflink.common.serialization import SimpleStringSchema

env = StreamExecutionEnvironment.get_execution_environment()
# 配置Kafka消费者
kafka_consumer = FlinkKafkaConsumer(
    topics='test_topic',
    deserialization_schema=SimpleStringSchema(),
    properties={'bootstrap.servers': 'localhost:9092', 'group.id': 'flink_group'}
)
# 添加数据源
stream = env.add_source(kafka_consumer)
# 数据处理：按空格分割单词，统计词频
result = stream \
    .flat_map(lambda line: line.split(" ")) \
    .map(lambda word: (word, 1)) \
    .key_by(lambda x: x[0]) \
    .sum(1)
# 输出到控制台
result.print()
# 执行任务
env.execute("Kafka Flink WordCount")

提交任务：flink run -c your_module_name your_script.py。

方案B：Apache Spark Streaming（微批处理，兼容批处理生态）

Spark Streaming通过微批处理模拟流处理，继承Spark的易用性与生态（如Hive、HDFS集成），适合已有Spark基础的场景。

安装Spark：下载稳定版（如3.2.1），解压后配置环境变量（~/.bashrc添加export SPARK_HOME=/usr/local/spark-3.2.1-bin-hadoop3.2、export PATH=$PATH:$SPARK_HOME/bin）。

编写Spark Streaming程序（Python示例）：使用KafkaUtils读取Kafka数据，进行实时单词计数：

from pyspark import SparkContext
from pyspark.streaming import StreamingContext
from pyspark.streaming.kafka import KafkaUtils

sc = SparkContext(appName="KafkaSparkStreaming")
ssc = StreamingContext(sc, batch_duration=1)  # 1秒批处理间隔
# 配置Kafka参数
kafka_params = {"metadata.broker.list": "localhost:9092"}
topics = ["test_topic"]
# 创建Kafka流
kafka_stream = KafkaUtils.createDirectStream(ssc, topics, kafka_params)
# 数据处理：提取值，按空格分割单词，统计词频
lines = kafka_stream.map(lambda x: x[1])
words = lines.flatMap(lambda line: line.split(" "))
word_counts = words.map(lambda word: (word, 1)).reduceByKey(lambda a, b: a + b)
# 输出结果
word_counts.pprint()
# 启动流处理
ssc.start()
ssc.awaitTermination()

提交任务：spark-submit --master local[2] your_script.py。

4. 数据存储与结果输出

处理后的实时数据可存储至以下系统，满足不同场景需求：

HDFS：作为Hadoop核心存储，适合长期存储海量数据（如通过Flink的FileSystemSink写入HDFS）；
关系型数据库：如MySQL、PostgreSQL，适合结构化数据查询（通过JDBC连接器写入）；
NoSQL数据库：如HBase、Redis，适合低延迟随机读写（如实时用户画像存储）。

5. 监控与优化

系统监控：使用Prometheus+Grafana监控Flink/Spark集群状态（如CPU、内存、任务延迟），或使用Flink Web UI、Spark Web UI查看任务详情；
性能调优：调整Kafka分区数（提高并行度）、Flink/Spark并行度（parallelism.default）、批处理间隔（Spark Streaming的batchDuration），优化GC策略（如使用G1GC减少停顿时间）。

通过以上步骤，可在Ubuntu环境下构建基于Hadoop生态的实时数据处理系统，结合流处理框架的高吞吐与低延迟特性，满足实时分析、监控等场景需求。