簡介

基本模組

• Apache Kafka
  訊息分發元件，資料採集後先入Kafka。

• Schema Registry
  Schema管理服務，訊息出入kafka、入hdfs時，給資料做序列化/反序列化處理。

• Kafka Connect
  提供kafka到其他儲存的管道服務，此次焦點是從kafka到hdfs，並建立相關HIVE表。

• Kafka Rest Proxy
  提供kafka的Rest API服務。

• Kafka Clients
  提供Client程式設計所需SDK。

說明：以上服務除Apache kafka由Linkedin始創並開源，其他元件皆由Confluent公司開發並開源。上圖解決方案由confluent提供。

基本邏輯步驟

• 資料通過Kafka Rest/Kafka Client寫入Kafka；
• kafka Connect任務作為consumer從kafka訂閱資料；
• kafka Connect任務建立HIVE表和hdfs檔案的對映關係；
• kafka connect任務收到資料後，以指定格式，寫入指定hdfs目錄;

實操：

1. 啟動服務

啟動kafka 服務

• 修改配置

/*
1.修改192.168.103.44、192.168.103.45、192.168.103.46三臺伺服器上配置
2.配置檔案中broker.id值分別修改為0、1、2
*/

cd  /home/ubuntu/confluent-2.0
 
.0
vi etc/kafka/server.properties

• 命令列啟動

cd  /home/ubuntu/confluent-2.0.0
nohup bin/kafka-server-start etc/kafka/server.properties &

• 服務說明
  kafka服務無Leader概念，服務訪問埠為9092

啟動Schema Register服務

• 命令列啟動

//wonderwoman叢集環境，只在woderwoman上啟動了服務
cd  /home/ubuntu/confluent-2.0.0
nohup bin/schema-registry-start etc/schema-registry
 
/schema-registry.properties &

• 服務說明
  Schema Register服務埠為8081

啟動Kafka Rest API服務

• 修改配置

// 修改schema伺服器地址和zookeeper伺服器地址
cd  /home/ubuntu/confluent-2.0.0
vi etc/kafka-rest/kafka-rest.properties

• 命令列啟動

//wonderwoman叢集環境，只在woderwoman上啟動了服務
cd  /home/ubuntu/confluent-2.0.0
nohup bin/kafka-rest-start ./etc/kafka-rest/kafka-rest.properties &

• 服務說明
  Schema Register服務埠為8082

啟動Kafka connect服務

• 修改配置

/*
修改如下配置項:
1.bootstrap.servers, 所有kafka broker的地址
2.group.id 標誌connect叢集，叢集內配一致
3.key.converter.schema.registry.url，schema服務地址
4.value.converter.schema.registry.url，schema服務地址
*/
cd /home/ubuntu/confluent-2.0.0/
vi etc/schema-registry/connect-avro-distributed.properties

• 命令列啟動

//啟動分散式connect，在zookeeper叢集內所有機器上啟動
cd /home/ubuntu/confluent-2.0.0/
nohup bin/connect-distributed etc/schema-registry/connect-avro-distributed.properties > /tmp/1.log

2. 建立schema

• Rest API方式
  
  • 準備shema的定義檔案，例如myschema.json
  • 註冊topic和schema，返回schema的ID

curl -X POST -i -H "Content-Type: application/vnd.schemaregistry.v1+json" --data @myschema.json http://localhost:8081/subjects/20160427/versions

• Kafka Client方式
  詳見 Kafka Client提供的介面，支援java/c/c++等語言

3. 建立Topic

建立的topic名稱為xianzhen，3個分割槽，每分割槽資料存兩份

• 命令列方式

cd /home/ubuntu/confluent-2.0.0/
./bin/kafka-topics --create --topic xianzhen --partitions 3 --replication-factor 2 --zookeeper 192.168.103.44:2181

• Rest API方式
  略，後續補充

• Kafka Client方式
  詳見 Kafka Client提供的介面，支援java/c/c++等語言

4. 建立Connector

分散式只支援以REST API的方式提交Connector作業

• Rest API方式
  
  • 建立配置檔案

vi confluentHive-connector.json
{
    "name": "hive-final",
    "config": {
        "connector.class": "io.confluent.connect.hdfs.HdfsSinkConnector",
        "tasks.max": "6",
        "topics": "confluentHive",
        "hdfs.url": "hdfs://ns1",
        "hadoop.conf.dir": "/home/ubuntu/hadoop-2.7.1/etc/hadoop",
        "hadoop.home": "/home/ubuntu/hadoop-2.7.1",
        "flush.size": "100",
        "rotate.interval.ms": "1000",
        "hive.integration":true,
        "hive.metastore.uris":"thrift://192.168.103.44:9083",
        "schema.compatibility":"BACKWARD",
        "hive.home":"/home/ubuntu/apache-hive-1.2.1-bin",
        "hive.conf.dir":"/home/ubuntu/spark-1.6.1/conf"
    }
}

• 部分配置說明

• 提交Connector

curl -X POST -H "Content-Type: application/json" --data @confluent-connector.json http://192.168.103.44:8083/connectors

5. 資料寫入kafka

寫入資料時，應帶上schema，或者已建立schema的ID

REST API方式

curl -X POST -H "Content-Type: application/vnd.kafka.avro.v1+json"   --data '{"value_schema_id":27, "records": [{"value": {"sip":"1.1.1.1", "dip":"1.1.1.2", "smac":"111", "dmac":"112"}}]}'  http://192.168.103.44:8082/topics/confluentHive

Kafka Client方式

詳見 Kafka Client提供的介面，支援java/c/c++等語言