前言：
本文章適用于在Windows上使用Flume 監(jiān)聽(tīng)Avro Client蕊程，模擬數(shù)據(jù)庫(kù)表的增量同步到Kafka中椒袍。首先確保你的flume-ng可以啟動(dòng)，跳過(guò)個(gè)別步驟可自行百度藻茂。
1驹暑、MySQL創(chuàng)建表：
DROP TABLE IF EXISTS `avro`;
CREATE TABLE `avro` (
`id` int(11) NOT NULL AUTO_INCREMENT,
`name` varchar(255) DEFAULT NULL,
`createdt` datetime DEFAULT NULL,
PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=11 DEFAULT CHARSET=latin1;
INSERT INTO `avro` VALUES ('1', 'a', '2018-11-21 04:00:00');
INSERT INTO `avro` VALUES ('2', 'b', '2018-11-22 05:00:00');
INSERT INTO `avro` VALUES ('3', 'c', '2018-11-23 06:00:00');
INSERT INTO `avro` VALUES ('4', 'd', '2018-11-24 07:00:00');
INSERT INTO `avro` VALUES ('5', 'e', '2018-11-25 08:00:00');
INSERT INTO `avro` VALUES ('6', 'f', '2018-11-26 09:00:00');
INSERT INTO `avro` VALUES ('7', 'g', '2018-11-27 10:00:00');
INSERT INTO `avro` VALUES ('8', 'h', '2018-11-28 11:00:00');
INSERT INTO `avro` VALUES ('9', 'i', '2018-11-29 12:00:00');
INSERT INTO `avro` VALUES ('10', 'j', '2018-11-30 13:56:41');
avro表如圖
image.png
2、Avro 的官網(wǎng)實(shí)例
2.1辨赐、創(chuàng)建Flume Avro Client ：(Thrift 同理)
打開(kāi)Eclipse 右擊
src/main/java 新建一個(gè)package org.flume.avro
新建Class MyApp.java
package org.flume.avro;
import org.apache.flume.Event;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.api.RpcClient;
import org.apache.flume.api.RpcClientFactory;
import org.apache.flume.event.EventBuilder;
import java.nio.charset.Charset;
public class MyApp {
public static void main(String[] args) {
// TODO Auto-generated method stub
MyRpcClientFacade client = new MyRpcClientFacade();
// Initialize client with the remote Flume agent's host and port
//端口與avro.conf a1.sources.r1.port一致
client.init("localhost", 41414);
String sampleData = "Hello Flume!";
for (int i = 0; i < 5; i++) {
client.sendDataToFlume(sampleData+" " + i);
}
System.out.println("輸出完畢");
client.cleanUp();
}
}
class MyRpcClientFacade {
private RpcClient client;
private String hostname;
private int port;
public void init(String hostname, int port) {
// Setup the RPC connection
this.hostname = hostname;
this.port = port;
this.client = RpcClientFactory.getDefaultInstance(hostname, port); //創(chuàng)建avro客戶端
// Use the following method to create a thrift client (instead of the above line):
// this.client = RpcClientFactory.getThriftInstance(hostname, port); //創(chuàng)建Thrift客戶端
}
public void sendDataToFlume(String data) {
// Create a Flume Event object that encapsulates the sample data
// 調(diào)用EventBuilder重載的withBody()方法优俘。
Event event = EventBuilder.withBody(data, Charset.forName("UTF-8"));
try {
client.append(event); // Send the event 發(fā)送數(shù)據(jù)
} catch (EventDeliveryException e) {
// clean up and recreate the client 清理并重新創(chuàng)建客戶端
client.close();
client = null;
client = RpcClientFactory.getDefaultInstance(hostname, port);
// Use the following method to create a thrift client (instead of the above line):
// this.client = RpcClientFactory.getThriftInstance(hostname, port);
}
}
public void cleanUp() {
// Close the RPC connection
client.close();
}
}
2.2、配置conf
Flume的conf目錄新建 avro.conf
a1.channels = c1
a1.sources = r1
a1.sinks = k1
a1.sources.r1.type = avro
a1.sources.r1.bind = 0.0.0.0
a1.sources.r1.port = 41414 //端口與MyApp.java中的port一致
a1.channels.c1.type = memory
a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink
a1.sinks.k1.topic = avrosrc
a1.sinks.k1.brokerList = localhost:9092
a1.sinks.k1.requiredAcks = 1
a1.sinks.k1.batchSize = 20
a1.sinks.k1.channel = c1
a1.sources.r1.channels = c1
2.3掀序、輸出到Kafka
此處省略Kafka啟動(dòng)步驟帆焕，詳見(jiàn)鏈接
新建Kafka Topic avrosrc
kafka-run-class.bat kafka.admin.TopicCommand --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic avrosrc
查看Topic avrosrc（此時(shí)為空）
kafka-console-consumer.bat --bootstrap-server localhost:9092 --topic avrosrc --from-beginning
啟動(dòng)flume-ng
D:\com\apache-flume-1.8.0-bin>flume-ng agent -c conf -f conf/avro.conf -n a1 -property "flume.root.logger=INFO,console"
Eclipse 運(yùn)行 MyApp.java（右鍵 → Run As → Java Application）
此時(shí)觀察 Topic 有數(shù)據(jù)進(jìn)入
image.png
3、Avro 自定義
每秒隨機(jī)讀取數(shù)據(jù)庫(kù)avro表的一條數(shù)據(jù)不恭，并輸出到Kafka叶雹，模擬增量數(shù)據(jù)
修改 MyApp.java
package org.flume.avro;
import org.apache.flume.Event;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.api.RpcClient;
import org.apache.flume.api.RpcClientFactory;
import org.apache.flume.event.EventBuilder;
import java.nio.charset.Charset;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.sql.Timestamp;
public class MyApp {
static final String DB_URL = "jdbc:mysql://localhost:3306/***"; //輸入DB名稱
static final String USER = "***"; //DB用戶名
static final String PASS = "***"; //DB密碼
public static void main(String[] args) {
// TODO Auto-generated method stub
MyRpcClientFacade client = new MyRpcClientFacade();
client.init("localhost", 41414);
Connection conn = null;
Statement stmt = null; //真實(shí)場(chǎng)景使用PreparedStatement防止SQL注入
try{
Class.forName("com.mysql.jdbc.Driver"); // 注冊(cè) JDBC 驅(qū)動(dòng)
conn = DriverManager.getConnection(DB_URL,USER,PASS); // 打開(kāi)鏈接
client.sendDataToFlume("Connect to db");
stmt = conn.createStatement(); // 執(zhí)行查詢
for(int i = 0;i < 10;i++){
int index = (int)(Math.random()*10) + 1;
String sql = "SELECT * FROM avro where id=" + index;
ResultSet rs = stmt.executeQuery(sql); // 保存到結(jié)果集
while(rs.next()){
int id = rs.getInt("id");
String name = rs.getString("name");
Timestamp createdt = rs.getTimestamp("createdt");
System.out.print("ID: " + id);
System.out.print(", 名稱: " + name);
System.out.print(", 創(chuàng)建時(shí)間: " + createdt);
System.out.print("\n");
//client.sendDataToFlume 發(fā)送數(shù)據(jù)！
client.sendDataToFlume("id: " + id + ", name: " + name + ", createdt: " + createdt);
}
rs.close();
try {
Thread.sleep(1000); //等待一秒,模擬增量場(chǎng)景
} catch (InterruptedException e) {
e.printStackTrace();
}
}
stmt.close();
conn.close();
}catch(SQLException se){ // 處理 JDBC 錯(cuò)誤
se.printStackTrace();
}catch(Exception e){ // 處理 Class.forName 錯(cuò)誤
e.printStackTrace();
}finally{ // 關(guān)閉資源
try{
if(stmt!=null) stmt.close();
}catch(SQLException se2){
}
try{
if(conn!=null) conn.close();
}catch(SQLException se){
se.printStackTrace();
}
}
client.sendDataToFlume("avro結(jié)束"); //測(cè)試中文是否亂碼：是
client.sendDataToFlume("avro over");
System.out.println("avro結(jié)束");
client.cleanUp();
}
}
class MyRpcClientFacade {
private RpcClient client;
private String hostname;
private int port;
public void init(String hostname, int port) {
// Setup the RPC connection
this.hostname = hostname;
this.port = port;
this.client = RpcClientFactory.getDefaultInstance(hostname, port); //創(chuàng)建avro客戶端
// Use the following method to create a thrift client (instead of the above line):
// this.client = RpcClientFactory.getThriftInstance(hostname, port); //創(chuàng)建Thrift客戶端
}
public void sendDataToFlume(String data) {
// Create a Flume Event object that encapsulates the sample data
// 調(diào)用EventBuilder重載的withBody()方法换吧。
Event event = EventBuilder.withBody(data, Charset.forName("UTF-8"));
try {
client.append(event); // Send the event 發(fā)送數(shù)據(jù)
} catch (EventDeliveryException e) {
// clean up and recreate the client 清理并重新創(chuàng)建客戶端
client.close();
client = null;
client = RpcClientFactory.getDefaultInstance(hostname, port);
// Use the following method to create a thrift client (instead of the above line):
// this.client = RpcClientFactory.getThriftInstance(hostname, port);
}
}
public void cleanUp() {
// Close the RPC connection
client.close();
}
}
再次運(yùn)行 MyApp.java
隨機(jī)讀取表中10條數(shù)據(jù)（每秒一條）折晦，輸出到Kafka
image.png
