目錄
1.系統(tǒng)架構(gòu)
2.環(huán)境搭建
2.1本地環(huán)境下kafka批量導(dǎo)入數(shù)據(jù)
2.2 kafka-manager的安裝與配置
3.1 Spark Streaming 性能調(diào)優(yōu)(一): 解決并行度
3.2 Spark Streaming 性能調(diào)優(yōu)(二): 解決task傾斜
上篇文章2.環(huán)境搭建的補(bǔ)充
其實下載下來的kafka中就自帶了zookeeper服務(wù), 進(jìn)行kafka/bin目錄下會看到有zookeeper-server-start.sh這個腳本,
執(zhí)行:
bin/zookeeper-server-start.sh config/zookeeper.properties
即可運(yùn)行與該kafka版本對應(yīng)的zookeeper
隨后執(zhí)行:
bin/kafka-server-start config/server.properties
執(zhí)行kafka
但是畢竟我們的spark程序是需要從kafka里面獲取數(shù)據(jù)的, 所以在本地環(huán)境下導(dǎo)入數(shù)據(jù)到kafka供我們的spark程序獲取是必須的.
網(wǎng)上很多人的帖子都是說用kafka-connector來導(dǎo)入數(shù)據(jù), 但是本地調(diào)試的話根本不需要這么大費(fèi)周章, 因為kafka提供了非常方便的數(shù)據(jù)導(dǎo)入腳本: kafka-console-producer
kafka-console-producer
輸入單挑數(shù)據(jù)
kafka-console-producer相當(dāng)于一個運(yùn)行在控制臺的生產(chǎn)者, 它可以由用戶手動一條一條地輸入數(shù)據(jù), 也可以直接指定一個數(shù)據(jù)文件批量導(dǎo)入, 用法也非常簡單:
bin/kafka-console-producer --broker-list localhost:9092 --topic test_topic
控制臺就會讀取用戶輸入, 一行為一條數(shù)據(jù), 如:
test123 hahaha
批量導(dǎo)入數(shù)據(jù)
如果想批量導(dǎo)入數(shù)據(jù), 我們則可以指定數(shù)據(jù)源文件
#file_path為數(shù)據(jù)源文件的路徑
bin/kafka-console-producer --broker-list localhost:9092 --topic test_topic < file_path
這樣稍等片刻數(shù)據(jù)就會導(dǎo)入完畢
一個有29w條數(shù)據(jù)的文件就導(dǎo)入好了!!!
均勻?qū)霐?shù)據(jù)
按照上面的步驟, kafka確實是導(dǎo)入了數(shù)據(jù), 但是你點(diǎn)進(jìn)去topic的視圖, 會發(fā)現(xiàn)數(shù)據(jù)都集中在一個partition上.這對于模擬數(shù)據(jù)源數(shù)據(jù)傾斜的情況非常有幫助, 但是如果我們需要的是數(shù)據(jù)均勻分布在不同partition上的情況呢?
kafka導(dǎo)入數(shù)據(jù)到分區(qū)的機(jī)制
要使數(shù)據(jù)均勻分布在kafka不同的分區(qū)上的話, 就必須了解kafka的分發(fā)數(shù)據(jù)的方式.
kafka默認(rèn)是使用隨機(jī)輪詢的方式來訪問不同的分區(qū), 并把數(shù)據(jù)寫入分區(qū)的, 而我們使用腳本kafka-console-producer.sh導(dǎo)入數(shù)據(jù)的時候, kafka是把這個文件中的記錄都當(dāng)做一個批次的數(shù)據(jù), 所以就只寫入一個分區(qū)了.
將文件分片寫入kafka
既然kafka把一個文件當(dāng)一個批次, 那么只要把文件分片成各個小的文件再導(dǎo)入就好了, 這里我的做法是編寫了一個shell腳本來分割大文件, 并將小文件逐個導(dǎo)入kafka.
#!/bin/bash
KAFKA_HOME=/home/lalala/CODE/Kafka/kafka_2.10-0.8.2.2
#文件路徑和文件名
file_path=$1
file_name=${file_path##*/}
file_dir=${file_path%/*}
echo "$file_path, $file_name, $file_dir"
#分區(qū)數(shù)
partitions=$2
let file_num=partitions*2
#數(shù)據(jù)源文件的行數(shù)
line_num=`sed -n '$=' $file_path`
echo "the line count of source file $file_name is $line_num"
#分片每個文件的行數(shù)
let chunk_line_num=line_num/file_num
echo "the line count of every chunk file is $chunk_line_num"
#創(chuàng)建存儲分片的文件夾
if [ -d "$file_dir/$file_name-chunks" ];then
echo "directory $file_dir/$file_name-chunks does exist."
else
mkdir "$file_dir/$file_name-chunks"
fi
#判斷文件是否創(chuàng)建成功,如果成功則將源文件內(nèi)容分片寫入
if [ -d "$file_dir/$file_name-chunks" ];then
echo "create dir $file_dir/$file_name-chunks success."
for((i=1;i<=$file_num;i++));
do
line_start=`expr \( $i - 1 \) \* $chunk_line_num + 1`
line_end=`expr $line_start + $chunk_line_num - 1`
awk "NR==$line_start,NR==$line_end" $file_path > "$file_dir/$file_name-chunks/$file_name-$i"
done
else echo "create dir $file_dir/$file_name-chunks fail."
fi
#將各個分片寫入kafka
for((i=1;i<=$file_num;i++));
do
su - $USER -c "JMX_PORT=10005 $KAFKA_HOME/bin/kafka-console-producer.sh --broker-list localhost:9092 --topic lala_topic < $file_dir/$file_name-chunks/$file_name-$i"
echo "$file_dir/$file_name-chunks/$file_name-$i imported"
done
exit 0
shell腳本的相關(guān)語法, 推薦各位做后端開發(fā)的童鞋還是都學(xué)習(xí)一下, 相信以后會大有幫助的.
