1. 分布式集群搭建概述在当今大数据处理领域构建一个完整的分布式数据处理平台已经成为企业级应用的标配。这次我要分享的是基于openEuler 24.03 LTS SP2操作系统整合Zookeeper、Hadoop、Spark、Kafka、Hive、Flume和MySQL等核心组件的集群搭建经验。这个方案特别适合需要处理海量数据但又不想投入过多硬件成本的中小型团队。选择openEuler作为基础操作系统有几个明显优势首先它是国产化发行版对国内用户更友好其次24.03 LTS SP2版本对大数据组件的兼容性经过特别优化最重要的是它的安全补丁更新及时适合生产环境使用。整套方案我实际测试下来在4台16核32GB内存的服务器上就能流畅运行TB级别的数据处理任务。2. 环境准备与系统配置2.1 硬件需求规划对于测试环境建议至少准备3台物理机或虚拟机1个Master2个Worker。每台机器配置建议CPU8核以上Spark和Hadoop都是CPU密集型内存32GB起步Hadoop和Spark都很吃内存存储500GB SSDHDFS和Kafka都需要高速磁盘网络万兆网卡最佳千兆也可用但性能会受限注意如果使用虚拟机务必关闭CPU的节能模式否则大数据组件的性能会大幅下降。2.2 openEuler系统初始化在所有节点上执行以下基础配置# 1. 设置主机名和hosts文件 hostnamectl set-hostname node1 # 根据实际修改 echo 192.168.1.101 node1 192.168.1.102 node2 192.168.1.103 node3 /etc/hosts # 2. 关闭防火墙和SELinux生产环境需谨慎 systemctl stop firewalld systemctl disable firewalld setenforce 0 sed -i s/SELINUXenforcing/SELINUXdisabled/g /etc/selinux/config # 3. 安装基础依赖 dnf install -y java-11-openjdk-devel python3 ssh pdsh # 4. 配置SSH免密登录 ssh-keygen -t rsa -P -f ~/.ssh/id_rsa cat ~/.ssh/id_rsa.pub ~/.ssh/authorized_keys chmod 600 ~/.ssh/authorized_keys2.3 JDK环境配置所有大数据组件都依赖Java环境建议使用OpenJDK 11# 设置JAVA_HOME环境变量 echo export JAVA_HOME/usr/lib/jvm/java-11-openjdk /etc/profile echo export PATH\$PATH:\$JAVA_HOME/bin /etc/profile source /etc/profile验证安装java -version # 应显示类似openjdk version 11.0.12 2021-07-20 LTS3. Zookeeper集群部署3.1 Zookeeper安装配置Zookeeper是整个分布式系统的协调者建议部署3节点集群# 在所有节点执行 wget https://downloads.apache.org/zookeeper/zookeeper-3.7.1/apache-zookeeper-3.7.1-bin.tar.gz tar -zxvf apache-zookeeper-3.7.1-bin.tar.gz -C /opt/ mv /opt/apache-zookeeper-3.7.1-bin /opt/zookeeper配置zoo.cfg# /opt/zookeeper/conf/zoo.cfg tickTime2000 initLimit10 syncLimit5 dataDir/var/lib/zookeeper clientPort2181 server.1node1:2888:3888 server.2node2:2888:3888 server.3node3:2888:3888在每个节点创建myid文件# 在node1上 echo 1 /var/lib/zookeeper/myid # 在node2上 echo 2 /var/lib/zookeeper/myid # 在node3上 echo 3 /var/lib/zookeeper/myid3.2 启动与验证# 所有节点启动 /opt/zookeeper/bin/zkServer.sh start # 查看状态 /opt/zookeeper/bin/zkServer.sh status # 应该看到一个leader和两个follower # 测试连接 /opt/zookeeper/bin/zkCli.sh -server node1:2181常见问题如果节点无法加入集群检查防火墙设置和myid文件是否正确。我曾遇到过因为myid文件权限问题导致选举失败的情况。4. Hadoop集群部署4.1 Hadoop安装配置下载Hadoop 3.3.4并解压wget https://downloads.apache.org/hadoop/common/hadoop-3.3.4/hadoop-3.3.4.tar.gz tar -zxvf hadoop-3.3.4.tar.gz -C /opt/ mv /opt/hadoop-3.3.4 /opt/hadoop配置环境变量echo export HADOOP_HOME/opt/hadoop /etc/profile echo export PATH\$PATH:\$HADOOP_HOME/bin:\$HADOOP_HOME/sbin /etc/profile source /etc/profile核心配置文件修改hadoop-env.shecho export JAVA_HOME/usr/lib/jvm/java-11-openjdk /opt/hadoop/etc/hadoop/hadoop-env.shcore-site.xmlconfiguration property namefs.defaultFS/name valuehdfs://node1:9000/value /property property namehadoop.tmp.dir/name value/opt/hadoop/tmp/value /property property nameha.zookeeper.quorum/name valuenode1:2181,node2:2181,node3:2181/value /property /configurationhdfs-site.xmlconfiguration property namedfs.replication/name value2/value /property property namedfs.namenode.name.dir/name value/opt/hadoop/data/namenode/value /property property namedfs.datanode.data.dir/name value/opt/hadoop/data/datanode/value /property /configurationyarn-site.xmlconfiguration property nameyarn.nodemanager.aux-services/name valuemapreduce_shuffle/value /property property nameyarn.resourcemanager.hostname/name valuenode1/value /property /configuration4.2 启动Hadoop集群在NameNode节点(node1)执行# 格式化HDFS hdfs namenode -format # 启动HDFS start-dfs.sh # 启动YARN start-yarn.sh验证集群# 检查进程 jps # 应有NameNode、DataNode、ResourceManager等进程 # 访问Web UI # NameNode: http://node1:9870 # ResourceManager: http://node1:8088避坑指南如果DataNode无法启动检查磁盘空间和权限。我曾遇到因为/tmp目录空间不足导致DataNode启动失败的情况。5. Spark集群部署5.1 Spark安装配置下载Spark 3.3.2 with Hadoop 3.3wget https://downloads.apache.org/spark/spark-3.3.2/spark-3.3.2-bin-hadoop3.tgz tar -zxvf spark-3.3.2-bin-hadoop3.tgz -C /opt/ mv /opt/spark-3.3.2-bin-hadoop3 /opt/spark配置环境变量echo export SPARK_HOME/opt/spark /etc/profile echo export PATH\$PATH:\$SPARK_HOME/bin /etc/profile source /etc/profile配置spark-env.shcp /opt/spark/conf/spark-env.sh.template /opt/spark/conf/spark-env.sh echo export JAVA_HOME/usr/lib/jvm/java-11-openjdk /opt/spark/conf/spark-env.sh echo export HADOOP_CONF_DIR/opt/hadoop/etc/hadoop /opt/spark/conf/spark-env.sh echo export SPARK_MASTER_HOSTnode1 /opt/spark/conf/spark-env.sh配置workers文件echo node1 node2 node3 /opt/spark/conf/workers5.2 启动Spark集群# 在Master节点(node1)启动 /opt/spark/sbin/start-all.sh验证集群# 检查进程 jps # 应有Master和Worker进程 # 访问Web UI # Spark Master: http://node1:8080提交测试任务spark-submit --class org.apache.spark.examples.SparkPi \ --master spark://node1:7077 \ /opt/spark/examples/jars/spark-examples_2.12-3.3.2.jar 100性能调优根据我的经验在spark-defaults.conf中调整以下参数可以提升性能spark.executor.memory 16g spark.executor.cores 4 spark.driver.memory 8g6. Kafka集群部署6.1 Kafka安装配置下载Kafka 3.3.1wget https://downloads.apache.org/kafka/3.3.1/kafka_2.12-3.3.1.tgz tar -zxvf kafka_2.12-3.3.1.tgz -C /opt/ mv /opt/kafka_2.12-3.3.1 /opt/kafka配置server.properties# /opt/kafka/config/server.properties broker.id1 # 每个节点唯一 listenersPLAINTEXT://:9092 advertised.listenersPLAINTEXT://node1:9092 # 根据实际节点修改 log.dirs/opt/kafka/logs zookeeper.connectnode1:2181,node2:2181,node3:2181 num.partitions3 default.replication.factor26.2 启动Kafka集群# 所有节点启动 /opt/kafka/bin/kafka-server-start.sh -daemon /opt/kafka/config/server.properties测试Kafka# 创建topic /opt/kafka/bin/kafka-topics.sh --create --bootstrap-server node1:9092 \ --replication-factor 2 --partitions 3 --topic test # 查看topic /opt/kafka/bin/kafka-topics.sh --list --bootstrap-server node1:9092 # 生产消息 /opt/kafka/bin/kafka-console-producer.sh --bootstrap-server node1:9092 --topic test # 另开终端消费消息 /opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server node1:9092 --topic test --from-beginning安全建议生产环境务必配置SASL认证我遇到过因为没配置认证导致Kafka被恶意写入的情况。可以参考Kafka官方文档配置SASL/PLAIN。7. Hive数据仓库部署7.1 Hive安装配置下载Hive 3.1.3wget https://downloads.apache.org/hive/hive-3.1.3/apache-hive-3.1.3-bin.tar.gz tar -zxvf apache-hive-3.1.3-bin.tar.gz -C /opt/ mv /opt/apache-hive-3.1.3-bin /opt/hive配置环境变量echo export HIVE_HOME/opt/hive /etc/profile echo export PATH\$PATH:\$HIVE_HOME/bin /etc/profile source /etc/profile配置hive-site.xmlconfiguration property namejavax.jdo.option.ConnectionURL/name valuejdbc:mysql://node1:3306/hive?createDatabaseIfNotExisttrue/value /property property namejavax.jdo.option.ConnectionDriverName/name valuecom.mysql.jdbc.Driver/value /property property namejavax.jdo.option.ConnectionUserName/name valuehive/value /property property namejavax.jdo.option.ConnectionPassword/name valuehive123/value /property property namehive.metastore.warehouse.dir/name value/user/hive/warehouse/value /property property namehive.exec.scratchdir/name value/tmp/hive/value /property /configuration7.2 MySQL安装配置# 在node1安装MySQL dnf install -y mysql-server mysql-connector-java systemctl start mysqld systemctl enable mysqld # 配置MySQL mysql_secure_installation # 创建Hive数据库和用户 mysql -u root -p CREATE DATABASE hive; CREATE USER hive% IDENTIFIED BY hive123; GRANT ALL PRIVILEGES ON hive.* TO hive%; FLUSH PRIVILEGES;将MySQL驱动复制到Hive lib目录cp /usr/share/java/mysql-connector-java.jar /opt/hive/lib/初始化Hive元数据库schematool -dbType mysql -initSchema7.3 启动Hive服务# 启动Metastore服务 hive --service metastore # 启动Hive CLI hive测试HiveCREATE TABLE test (id INT, name STRING); INSERT INTO test VALUES (1, test); SELECT * FROM test;性能优化Hive查询慢是常见问题我通常会调优这些参数set hive.exec.paralleltrue; set hive.exec.parallel.thread.number16; set mapreduce.map.memory.mb4096;8. Flume日志收集系统部署8.1 Flume安装配置下载Flume 1.11.0wget https://downloads.apache.org/flume/1.11.0/apache-flume-1.11.0-bin.tar.gz tar -zxvf apache-flume-1.11.0-bin.tar.gz -C /opt/ mv /opt/apache-flume-1.11.0-bin /opt/flume配置环境变量echo export FLUME_HOME/opt/flume /etc/profile echo export PATH\$PATH:\$FLUME_HOME/bin /etc/profile source /etc/profile创建Flume配置文件flume-kafka.conf# 定义agent的sources、channels、sinks agent.sources logSource agent.channels memChannel agent.sinks kafkaSink # 配置source agent.sources.logSource.type exec agent.sources.logSource.command tail -F /var/log/messages agent.sources.logSource.channels memChannel # 配置channel agent.channels.memChannel.type memory agent.channels.memChannel.capacity 10000 agent.channels.memChannel.transactionCapacity 1000 # 配置sink agent.sinks.kafkaSink.type org.apache.flume.sink.kafka.KafkaSink agent.sinks.kafkaSink.kafka.bootstrap.servers node1:9092,node2:9092,node3:9092 agent.sinks.kafkaSink.kafka.topic logs agent.sinks.kafkaSink.channel memChannel8.2 启动Flumeflume-ng agent --conf /opt/flume/conf --conf-file /opt/flume/conf/flume-kafka.conf \ --name agent -Dflume.root.loggerINFO,console验证数据流# 在Kafka消费者查看日志 /opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server node1:9092 --topic logs稳定性建议Flume容易因为日志文件轮转导致数据丢失建议使用spooldir source而不是exec source。我在生产环境中吃过这个亏。9. 集群集成与验证9.1 组件集成测试验证整个数据流水线Flume收集日志到KafkaSpark Streaming从Kafka消费数据处理结果存入HDFSHive创建外部表分析数据Spark Streaming示例代码from pyspark.sql import SparkSession from pyspark.sql.functions import * spark SparkSession.builder \ .appName(KafkaStreamExample) \ .getOrCreate() df spark.readStream \ .format(kafka) \ .option(kafka.bootstrap.servers, node1:9092) \ .option(subscribe, logs) \ .load() query df.selectExpr(CAST(value AS STRING)) \ .writeStream \ .outputMode(append) \ .format(parquet) \ .option(path, /user/spark/logs) \ .option(checkpointLocation, /user/spark/checkpoint) \ .start() query.awaitTermination()9.2 性能监控建议推荐监控方案Prometheus Grafana监控集群指标ELK收集和分析日志Hadoop自带的Web UI监控HDFS和YARNSpark History Server分析Spark作业关键监控指标HDFS存储使用率YARN资源利用率Kafka积压消息数Spark任务执行时间10. 常见问题与解决方案10.1 安装部署问题问题现象可能原因解决方案Zookeeper无法选举防火墙未关闭/myid文件错误检查端口2181/2888/3888是否开放验证myid文件DataNode无法启动磁盘空间不足/权限问题检查df -h输出确保HDFS数据目录有足够空间和正确权限Spark Worker不注册网络不通/主机名解析错误检查ping和hostname命令确保所有节点能互相解析10.2 运行时报错处理HDFS写入失败# 报错Could only write X replicas instead of minReplication # 解决 hdfs dfsadmin -setrep 2 /path/to/fileYARN资源不足# 报错Container killed by YARN for exceeding memory limits # 解决调整Spark内存参数 spark-submit --executor-memory 8G ...Kafka消息积压# 查看积压情况 /opt/kafka/bin/kafka-consumer-groups.sh --bootstrap-server node1:9092 \ --describe --group your_group # 解决方案增加消费者或分区数10.3 性能优化经验HDFS调优!-- hdfs-site.xml -- property namedfs.datanode.handler.count/name value20/value /propertySpark SQL优化-- 使用分区表 CREATE TABLE logs (msg STRING) PARTITIONED BY (dt STRING); -- 启用动态分区 SET hive.exec.dynamic.partitiontrue; SET hive.exec.dynamic.partition.modenonstrict;Kafka生产端优化batch.size16384 linger.ms5 compression.typesnappy这套分布式集群搭建方案经过我多次实践验证能够稳定支持TB级数据处理需求。根据实际业务场景可以灵活调整组件配置和规模。比如对于实时性要求高的场景可以增加Kafka分区和Spark Streaming的并行度对于批处理任务则可以调大Hadoop和Spark的堆内存配置。