1. 大數據領域數據類型
1.1 有界數據
一般批處理(一個文件 或者一批文件),不管文件多大,都是可以度量
mapreduce hive sparkcore sparksql
1.2 無界數據
源源不斷的流水一樣 (流數據)
Storm SparkStreaming
2. 消息隊列(Message Queue)
- 消息 Message
- 網絡中的兩臺計算機或者兩個通訊設備之間傳遞的數據,例如說:文本、音樂、視頻等內容
- 隊列 Queue
- 一種特殊的線性表(數據元素首尾相接),特殊之處在於只允許在首部移除元素和在尾部追加元素。入隊、出隊。
- 消息隊列 MQ
- 消息+隊列
- 保存消息的隊列
- 消息的傳輸過程中的容器
- 主要提供生產、消費接口供外部調用做數據的存儲和獲取
3. 消息隊列的分類
3.1 點對點(P2P)
- 一個生產者生產的消息只能被一個消費者消費
3.2 發佈訂閱(Pub/Sub)
消息隊列(Queue)、主題(Topic)、發佈者(Publisher)、訂閱者(Subscriber)
- 消息的發佈者
- 消息的訂閱者
- 每個消息可以有多個消費者,彼此互不影響。比如我發佈一個微博:關注我的人都能夠看到。
4. Kafka的簡介
- 在大數據領域呢,為了滿足日益增長的數據量,也有一款可以滿足百萬級別消息的生成和消費,分佈式、持久穩定的產品——Kafka
- Kafka是分佈式的發佈—訂閱消息系統(基於PS的一個消息隊列)
- 它最初由LinkedIn(領英)公司發佈,使用Scala語言編寫
- Kafka是一個高吞吐量的、持久性的、分佈式發佈訂閱消息系統
- 它主要用於處理活躍的數據(登錄、瀏覽、點擊、分享、喜歡等用戶行為產生的數據
5. Kafka的特點
- 高吞吐量
- 可以滿足每秒百萬級 別消息的生產和消費(生產消費 )
- 持久性
- 有一套完善的消息存儲機制,確保數據的高效安全的持久化 (數據的存儲)
- 分佈式
- 基於分佈式的擴展和容錯機制;Kafka的數據都會複製到幾臺服務器上。當某一臺故障失效時,生產者和消費者轉而使用其它的機器——整體健壯性
6. Kafka的組件
- 一個消息隊列需要哪些部分?
- 生產
- 消費
- 消息類別
- 存儲等等
- Topic(主題)
- Kafka處理的消息的不同分類
- Broker (消息代理)
- Kafka集群中的一個kafka服務節點稱為一個broker,主要存儲消息數據,存在硬盤中。每個topic都是有分區的
- Partition (物理上的分區)
- 一個topic在broker中被分為1個或者多個partition,分區在創建topic的時候指定
- Message (消息)
- 消息,是通信的基本單位,每個消息都屬於一個partition
7. Kafka的服務
- Producer : 消息和數據的生產者,向Kafka的一個topic發佈消息
- Consumer :消息和數據的消費者,定於topic並處理其發佈的消息
- Zookeeper :協調kafka的正常運行
8. Kafka的安裝
8.1 單機版的安裝
- 準備kafka
- kafka_2.10-0.10.0.1.tgz
- 解壓kafka
- tar -zxvf kafka_2.10-0.10.0.1.tgz -C /opt/
- 重命名
- mv kafka_2.10-0.10.0.1.tgz kafka
- 配置環境變量
export KAFKA_HOME=/opt/kafka
export PATH=$PATH:$KAFKA_HOME/bin
- 編輯server.properties
broker.id=1
log.dirs=/opt/kafka/logs
zookeeper.connect=uplooking03:2181,uplooking04:2181,uplooking05:2181
listeners=PLAINTEXT://:9092
- 啟動kafka-server服務
kafka-server-start.sh [-daemon] server.properties
- 停止kafka服務
kafka-server-stop.sh
8.2 集群的安裝
只需要在每個機器上修改對應的 ==broker.id=1== 即可
9. Kafka中Topic的操作
- 創建topic
kafka-topics.sh --create --topic t1 --partitions 3 --replication-factor 1 --zookeeper uplooking03:2181,uplooking04:2181
- ==注意: 創建topic過程的問題,replication-factor個數不能超過brokerserver的個數==
- 查看topic
kafka-topics.sh --list --zookeeper uplooking03
- 查看具體topic的詳情
kafka-topics.sh --describe --topic t1 --zookeeper uplooking04:2181
PartitionCount:topic對應的partition的個數
ReplicationFactor:topic對應的副本因子,說白就是副本個數
Partition:partition編號,從0開始遞增
Leader:當前partition起作用的breaker.id
Replicas: 當前副本數據存在的breaker.id,是一個列表,排在最前面的其作用
Isr:當前kakfa集群中可用的breaker.id列表
- 修改topic(不能修改replication-factor,以及只能對partition個數進行增加,不能減少 )
kafka-topics.sh --alter --topic t1 --partitions 4 --zookeeper uplooking03
- 刪除Topic
kafka-topics.sh --delete --topic t1 --zookeeper uplooking03
- ps:這種刪除只是標記刪除,要想徹底刪除必須設置一個屬性,在server.properties中配置delete.topic.enable=true,否則只是標記刪除
- 配置完成之後,需要重啟kafka服務
10. Kafka中的生產者和消費者接口
- 自己寫代碼實現kafka提供的消息生產和消費的接口
- kafka自身也實現了自身的生產和消費的接口,給出了兩個工具(kafka-console-producer.sh , kafka-console-consumer.sh)
11. Kafka自帶的生產和消費消息的工具
11.1 kafka-console-producer.sh(生產工具)
kafka-console-producer.sh --topic t1 --broker-list uplooking03:9092,uploo
king04:9092,uplooking05:9092
11.2 kafka-console-consumer.sh(消費工具)
kafka-console-consumer.sh --zookeeper uplooking03 --topic t1
--from-beginning:從頭開始消費
--blacklist:黑名單過濾(kafka-console-consumer.sh --zookeeper uplooking03 --blacklist t1,t3)
--whitelist:白名單過濾(kafka-console-consumer.sh --zookeeper uplooking03 --whitelist t2)
ps:--topic|--blacklist|--whitelist 只能出現其中一個
12. ==Flume與Kafka的整合==
- 配置flume的agent配置文件
- touch flume-kafka.properties
# 對各個組件的描述說明
# 其中a1為agent的名字
# r1是a1的source的代號名字
# c1是a1的channel的代號名字
# k1是a1的sink的代號名字
############################################
a1.sources = r1
a1.sinks = k1
a1.channels = c1
# 用於描述source的,類型是netcat網絡
a1.sources.r1.type = netcat
# source監聽的網絡ip地址和端口號
a1.sources.r1.bind = uplooking01
a1.sources.r1.port = 44444
# 用於描述sink,類型是kafka
a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink
a1.sinks.k1.topic = hadoop
a1.sinks.k1.brokerList = uplooking03:9092,uplooking04:9092,uplooking05:9092
a1.sinks.k1.requiredAcks = 1
a1.sinks.k1.batchSize = 2
# 用於描述channel,在內存中做數據的臨時的存儲
a1.channels.c1.type = memory
# 該內存中最大的存儲容量,1000個events事件
a1.channels.c1.capacity = 1000
# 能夠同時對100個events事件監管事務
a1.channels.c1.transactionCapacity = 100
# 將a1中的各個組件建立關聯關係,將source和sink都指向了同一個channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
- 啟動flume開始採集數據
[root@uplooking01:/opt/flume/conf]
flume-ng agent --name a1 --conf-file flume-kafka.properties
- 開啟Kafka消息消費工具
[root@uplooking03:/opt/flume/conf]
kafka-console-consumer.sh --zookeeper uplooking03 --topic hadoop
- 給flume監聽的Source發送數據
[root@uplooking03:/]
nc uplooking01 44444
- 現在就可以到kafka的消費工具(kafka-console-consumer.sh)中區查看nc發送的數據
13. Kafka的API操作(生產者和消費者)
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka_2.10</artifactId>
<version>0.10.0.1</version>
</dependency>
13.1 Kafka的生產者
- 創建生產者的配置文件 producer.properties
bootstrap.servers=uplooking03:9092,uplooking04:9092,uplooking05:9092
key.serializer=org.apache.kafka.common.serialization.StringSerializer
value.serializer=org.apache.kafka.common.serialization.StringSerializer
- 創建生產者並且發送數據到topic中
public class MyKafkaProducer {
public static void main(String[] args) throws IOException {
Properties prop = new Properties();
prop.load(MyKafkaProducer.class.getClassLoader().getResourceAsStream("producer.properties"));
KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(prop);
kafkaProducer.send(new ProducerRecord<String, String>("hadoop", "name", "admin123"));
kafkaProducer.close();
}
}
13.2 Kafka的消費者
- 創建消費者的配置文件consumer.properties
zookeeper.connect=uplooking03:2181,uplooking04:2181,uplooking05:2181
group.id=test-consumer-group
bootstrap.servers=uplooking03:9092,uplooking04:9092,uplooking05:9092
key.deserializer=org.apache.kafka.common.serialization.StringDeserializer
value.deserializer=org.apache.kafka.common.serialization.StringDeserializer
- 創建消息消費者消費topic中的數據
public static void main(String[] args) throws Exception {
Properties prop = new Properties();
prop.load(MyKafkaConsumer.class.getClassLoader().getResourceAsStream("consumer.properties"));
KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<String, String>(prop);
Collection topics = new ArrayList();
topics.add("hadoop");
kafkaConsumer.subscribe(topics);
while (true) {
ConsumerRecords<String, String> records = kafkaConsumer.poll(1000);
for (ConsumerRecord<String, String> record : records) {
System.out.println(record.value());
}
}
}
- 自定義分區(MyCustomPartition)
package com.uplooking.bigdata.kafka.partition;
public class MyCustomPartition implements Partitioner {
public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
//獲取分區數, 分區編號一般都是從0開始
int partitionSize = cluster.partitionCountForTopic(topic);
int keyHash = Math.abs(key.hashCode());
int valueHash = Math.abs(value.hashCode());
return keyHash % partitionSize;
}
public void close() {
}
public void configure(Map<String, ?> configs) {
}
}
- 配置自定義分區(producer.properties)
partitioner.class=com.uplooking.bigdata.kafka.partition.MyCustomPartition
end:如果你覺得本文對你有幫助的話,記得點贊轉發,你的支持就是我更新動力。
相關推薦
推薦中...