hadoop、zk、kafka命令集合

#Named
a1.sources = r1
a1.channels = c1 
a1.sinks = k1#Source   注意kafkasource会给我们封装even的header的timestamp属性时间,不使用系统时间
#因为上传也消耗时间,所以,可能会产生误差,而使用数据本身ts时间,则能实现精准上传
a1.pe = org.apache.flume.source.kafka.KafkaSource
#只是为了防止112的kafka故障,从而能从113进入kafka
a1.sources.r1.kafka.bootstrap.servers = hadoop112:9092,hadoop113:9092
a1.sources.up.id = gmall
a1.sources.pics = topic_log
a1.sources.r1.batchSize = 1000
#若batchsize没有装满,则最大等10s传送
a1.sources.r1.batchDurationMillis = 1000
a1.sources.r1.useFlumeEventFormat = false #Interceptor
a1.sources.r1.interceptors = i1
a1.sources.r1.pe = com.ck.LogHeaderTimeStampInterceptor.LogHeaderTimeStampInterceptor$MyBuilder#Channel 
a1.pe = file
#将source传来的数据写入filechannel
a1.channels.c1.dataDirs = /opt/module/flume-1.9.0/jobs/filechannel
#维护被sink拿走的数据(也就是被sink拿走后,数据被标记,这些标记存在checkpoint)
a1.channels.c1.checkpointDir = /opt/module/flume-1.9.0/jobs/checkpoint
#Checkpoints是否备份维护数据(checkpoint),若未true,则backupCheckpointDir为备份路径在otherdisk里面
#a1.channels.c1.useDualCheckpoints= true 
#a1.channels.c1.backupCheckpointDir = /otherdisk 
#若source往channel放不下数据适合,会发生回滚事务(把source放进去的数据也抛弃,重新source该批数据).
#接上面,设置5s则为在5s内一直尝试放入,而不发生回滚,当超过5s再发生回滚
a1.channels.c1.keep-alive = 5 #Sink 
a1.pe = hdfs
a1.sinks.k1.hdfs.path = /origin_data/gmall/log/topic_log/%Y-%m-%d
#是否使用本地时间戳
a1.sinks.k1.hdfs.useLocalTimeStamp = false 
a1.sinks.k1.hdfs.filePrefix = log-
#是否按照时间滚动文件夹
a1.sinks.und = false
#多久生成一个新的文件 (若10s内未上传完毕,则hdfs创建一个新文件夹收集继续上传内容)
a1.sinks.llInterval = 10
#设置每个文件的滚动大小
a1.sinks.llSize = 134217728
#文件的滚动与Event数量无关
a1.sinks.llCount = 0
#设置文件类型，可支持压缩
a1.sinks.k1.hdfs.fileType = CompressedStream
a1.sinks.deC = lzop#Bind  
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1 #运行
flume-ng agent -c $FLUME_HOME/conf -f $FLUME_HOME/jobs/f -n a1 -logger=INFO,console

f2脚本

#!/bin/bash
if [ $# -lt 1 ]
then
echo "请输入参数	 {start|stop}"
exit
fifor i in hadoop114
do
case $1 in 
start)
ssh $i "nohup flume-ng agent -c $FLUME_HOME/conf -f $FLUME_HOME/jobs/f -n a1 -logger=INFO,console 1>/opt/module/flume-1.9.0/logs/flume.log  2>&1 &"
;;
stop)
ssh $i " ps -ef | f | grep -v grep | awk '{print $2}' | xargs -n1 kill -9"
;;
*)
echo "参数错误,请输入参数	 {start|stop}"
exit
;;
esac
done

4 Kafka命令

4.1 单点启动

-- 启动
[atguigu@hadoop112 kafka_2.11-2.4.1]$ kafka-server-start.sh -daemon config/server.properties-- 停止
[atguigu@hadoop112 kafka_2.11-2.4.1]$ kafka-server-stop.sh

4.2 topic-命令行操作

1)查看所有的Topic
[atguigu@hadoop112 kafka_2.11-2.4.1]$kafka-topics.sh --list --bootstrap-server hadoop112:90922)创建Topic    topic 本次创建topic名称  partitions分区个数  replication-factor副本个数 3
[atguigu@hadoop112 kafka_2.11-2.4.1]$ kafka-topics.sh --create --bootstrap-server hadoop112:9092 --topic second --partitions 2 --replication-factor 33) 查看topic的详情
[atguigu@hadoop112 kafka_2.11-2.4.1]$ kafka-topics.sh --describe --bootstrap-server hadoop112:9092 --topic second4) 修改Topic 注意:只能增加分区不能减少分区
[atguigu@hadoop112 kafka_2.11-2.4.1]$ kafka-topics.sh --alter --bootstrap-server hadoop112:9092 --topic second --partitions 35) 删除Topic
[atguigu@hadoop112 kafka_2.11-2.4.1]$ kafka-topics.sh --delete --bootstrap-server hadoop112:9092 --topic third

注意
– 查看topic的详情时

Topic: second	PartitionCount: 2	ReplicationFactor: 3	Configs: segment.bytes=1073741824Topic: second	Partition: 0	Leader: 3	Replicas: 3,2,4	Isr: 3,2,4Topic: second	Partition: 1	Leader: 4	Replicas: 4,3,2	Isr: 4,3,2Topic: 名字	PartitionCount: 分区个数	ReplicationFactor:副本数	
Topic: 名字	Partition: 分区序号		Leader: L在那个broker	Replicas: R在那几个broker	
Isr:活着的R在那几个broker上 

4.3 命令行操作-生产和消费

1) 生产者
[atguigu@hadoop112 kafka_2.11-2.4.1]$ kafka-console-producer.sh --broker-list hadoop112:9092 -topic second2) 消费者(broker内分区无序,但分区内数据有序)
[atguigu@hadoop112 kafka_2.11-2.4.1]$ kafka-console-consumer.sh --bootstrap-server hadoop112:9092 -topic second
-- 新组从头开始读取数据	
[atguigu@hadoop112 kafka_2.11-2.4.1]$ kafka-console-consumer.sh --bootstrap-server hadoop112:9092 -topic second --from-beginning3)消费者组(offset 的 g[组]-t[主题]-p[分区])
-- 在config目录下修改配置文件consumer.properties内的group.id=g112,并在创建时候指定该配置文件
[atguigu@hadoop112 kafka_2.11-2.4.1]$ kafka-console-consumer.sh --bootstrap-server hadoop112:9092 -topic second --group g11

消费者注意

如果默认创建消费者:而不指定组,则默认分配组,每次默认创建的组都会不同kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic second
注意:若同一组的消费者数超过分区数,则会自动挤掉至和分区相同的消费者.