您现在的位置是:首页 >技术交流 >项目开发经验网站首页技术交流
项目开发经验
hadoop
1.namenode中有专门的工作线程池用于处理与datanode的心跳信号 dfs.namenode.handler.count=20 * log2(Clust
2.编辑日志存储路径 dfs.namenode.edits.dir 设置与镜像文件存储路径 dfs.namenode分开存放,可以达到提高并发
3.yarn参数调优,单个服务节点上yarn可获得的默认最大内存为8G,而单个任务默认可申请最大内存为8G;
4.hdfs和硬盘使用控制在70%以下。
5.hadoop宕机: 5.1 MR过载时,应控制yarn,任务运行的数量,和每个任务申请的 最大内存。调整参数:yarn.scheduler.maximum-allocation-mb(单个任务可申请的最多物 理内存量,默认是 8192MB)
5.2 载入文件过大造成namenode宕机时,应使用kafka做缓存(调高),控制kafka写入hdfs的速度,在高峰期过后数据同步自动跟上。
flume
1.Flume内存配置4G(flume-env.sh中配置)
FileChannel优化
/etc/flume-ng/conf
目录下配置文件存放路径
2.datadir目录指向多个路径,配置到不同硬盘的多个目录下,可以提高并发。checkpointDir和backupcheckpointDir配置到不同硬盘的对应目录下,以便宕机时快速启用备份。
agent.channels.channel1.dataDirs = /path/to/data/dir1, /path/to/data/dir2, /path/to/data/dir3
agent.channels.channel1.checkpointDir = /path/to/checkpoint/dir agent.channels.channel1.backupCheckpointDir = /path/to/backup/checkpoint/dir
3.hdfs sink 小文件处理,控制滑动窗口溢写3个参数,hdfs.rollInterval 、 hdfs.rollSize 、 hdfs.rollCount
kafka
kafka吞吐量测试
kafka内存6G(不能超过6G)
kafka 每天数据1亿条,1150条每秒
kafka 消费能力不足怎么办?
1.并发 增加kafka topic的分区数以及对应的消费者数 (两者统一,缺一不可)
2.频率 拉取的速度需要跟得上生产的速度,如果下游处理数据不及时,可以提高每批次拉取的数据量
kafka挂掉期间了数据怎么办?
内部:
1.重启kafka
2.按消费者偏移量回到发生宕机之前的位置重新开始处理消息
3.kafka的日志和备份数据
外部:
1.flume 的channel 短期可以缓存
2.日志服务器的备份,可以等kafka重启后再次加载
kafka数据重复:在下一级消费者中去重(redis、spark streaming、hive dwd层)
Hive
1.自定义函数
UDF 继承extendsUDF类,重写evaluate(),实现重定义日的功能
UDTF 继承GUDTF类,重写 initialize() process()close()
UDAF
2.优化
2.1 存储
hdfs 上存储格式、列式存储对数仓的优化提升比较明显,提升压缩效率
小文件处理,归档合并小文件
2.2计算
1针对不同计算引擎依赖hive元数据信息生成更优的执行计划,
2mr计算的时候map阶段可以提前合并小文件再切片,减少数据切斜
3环形缓冲区落盘阶段可以提前聚合,减少io
4开启JVM重用
mysql元数据备份问题
重点:可能导致整个集群无法运行,至少保证每日零点后备份到其它服务器两个副本
tez的优点: 可以将多个有依赖的作业转换成一个大作业,减少IO,只写入一次HDFS,提高
计算性能
sqoop:
1.sqoop 的parquet数据从hdfs导出到mysql,需先转换成text
2.Hive 中的 Null 在底层是以“N”来存储,而 MySQL 中的 Null 在底层就是 Null,为了 保证数据两端的一致性。在导出数据时采用--input-null-string 和--input-null-non-string 两 个参数。导入数据时采用--null-string 和--null-non-string。
sqoop import
--connect jdbc:mysql://hostname:port/database
--username username
--password password
--table table_name
--null-string '\N'
--null-non-string '\N'
--target-dir /path/to/output_dir
sqoop 数据一致性问题:
使用多个map如何保证? 使用—staging-table –clear-staging 任务执行成功首先在 tmp 临时表中,然后将 tmp 表中的数据复制到目标表中(这个时 候可以使用事务,保证事务的一致性
sparkstreaming优雅关闭,监听标志位,并使用ssc.stop()
-
定义一个标志位:定义一个标志位(如
isStopped
),用于控制应用程序的关闭状态。 -
监听流式数据源的结束:如果你的流式数据源有一个结束信号,比如 Kafka 的消费者消费完所有消息,你可以在监听到数据源结束时将标志位设为
true
。
import org.apache.spark.SparkConf;
import org.apache.spark.streaming.Duration;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
public class StreamingApp {
public static void main(String[] args) throws InterruptedException {
// 创建 SparkConf 对象
SparkConf conf = new SparkConf().setAppName("StreamingApp").setMaster("local[2]");
// 创建 StreamingContext,并设置批处理间隔
JavaStreamingContext streamingContext = new JavaStreamingContext(conf, new Duration(5000));
// 定义标志位
final boolean[] isStopped = {false};
// 注册 JVM 钩子
Runtime.getRuntime().addShutdownHook(new Thread() {
@Override
public void run() {
System.out.println("Shutting down the application...");
// 设置标志位为 true
isStopped[0] = true;
// 停止 StreamingContext
streamingContext.stop(true, true);
System.out.println("Application is successfully shut down.");
}
});
// 从流式数据源读取数据,并进行处理
// ...
// 控制循环退出条件
while (!isStopped[0]) {
// 处理流式数据
// ...
}
// 停止 StreamingContext
streamingContext.stop();
// 关闭 Spark 上下文
streamingContext.sparkContext().stop();
}
}