目錄

目錄

一、序言

驗證資料

一、序言

實驗用到的元件有：docker、kafka、kafka-manager、zookeeper、flume；由於資源限制使用docker下安裝kafka和zookeeper，在試驗機上直接安裝flume和kafka-manager。

實驗內容：1.本地產生日誌資料，通過log4j將日誌收集到flume中，flume將資料sink到kafka中；2.flume從kafka中獲取資料然後列印到控制檯中；（或者使用flink從kafka中拿到資料，新增標識欄位後重新放入kafka另一個topic中，注稍後補全這部分）

實驗目的：通過實驗學習到docker安裝、使用、kafka操作、flume操作以及部署工作；

所需maven依賴包

  <!-- kafka start -->
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka_2.11</artifactId>
            <version>0.9.0.1</version>
        </dependency>

        <!-- kafka end -->

        <!--flume-->
        <dependency>
            <groupId>org.apache.flume.flume-ng-clients</groupId>
            <artifactId>flume-ng-log4jappender</artifactId>
            <version>1.7.0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flume</groupId>
            <artifactId>flume-ng-core</artifactId>
            <version>1.7.0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flume</groupId>
            <artifactId>flume-ng-configuration</artifactId>
            <version>1.7.0</version>
        </dependency>

        <!--flink-->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-java</artifactId>
            <version>1.5.0</version>
        </dependency>
        <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-connector-kafka-0.9 -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-kafka-0.9_2.11</artifactId>
            <version>1.5.0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-java_2.11</artifactId>
            <version>1.5.0</version>
        </dependency>




        <!-- log4j start -->
        <dependency>
            <groupId>log4j</groupId>
            <artifactId>log4j</artifactId>
            <version>1.2.17</version>
        </dependency>
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-api</artifactId>
            <version>1.7.5</version>
        </dependency>
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-log4j12</artifactId>
            <version>1.7.5</version>
        </dependency>
        <!-- log4j end -->

        <!-- https://mvnrepository.com/artifact/com.alibaba/fastjson -->
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>1.2.47</version>
        </dependency>
 

二、環境準備

2.1 docker環境準備

實驗機環境為centos7

2.2 安裝zookeeper、kafka、kafka-manager環境

2.2.1 zookeeper

使用docker search zookeeper命令獲取資源庫zookeeper列表

下載首先pull獲取 wurstmeister的zookeeper

啟動zookeeper：docker run -d --name zookeeper -p 2181:2181 -t wurstmeister/zookeeper

2.2.2 kafka

使用docker search zookeeper命令獲取資源庫kafka列表

下載首先pull獲取 wurstmeister的kafka

啟動kafka：

docker run -d --name kafka -p 9092:9092  -e KAFKA_BROKER_ID=0  -e KAFKA_ZOOKEEPER_CONNECT=192.168.83.112:2181  -e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://192.168.83.112:9092  -e KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:9092 -t wurstmeister/kafka

解釋：

KAFKA_BROKER_ID=0               //broker id,如果想要啟動多個就執行多次命令保證 id不相同就行了

KAFKA_ZOOKEEPER_CONNECT=192.168.83.112:2181  //外界連線kafka所需

KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://192.168.83.112:9092  //外界連線kafka所需，地址是宿主機地址

KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:9092

2.2.3 kafka-manager

安裝在宿主機上，沒有安裝到docker中，因為docker中的映象存在元件缺失

從 GitHub 下載後編譯，編譯方法自行百度，或者直接下載已經編譯好的

連結：https://pan.baidu.com/s/1zmhG6-eP_0RsGDxvcEMzyw

密碼：sc8w

解壓到最終安裝位置，然後配置兩項

kafka-manager.zkhosts="192.168.83.112:2181
akka {
  loggers = ["akka.event.slf4j.Slf4jLogger"]
  loglevel = "INFO"
  logger-startup-timeout = 30s
}

然後執行啟動命令

 nohup bin/kafka-manager -Dconfig.file=conf/application.conf -Dhttp.port=9000 &

2.3 安裝flume

從官網下載檔案，解壓後上傳到最終安裝目錄，

配置java路徑到flume-env.sh

export JAVA_HOME=/usr/java/jdk1.8.0_171-amd64

2.4 安裝flink

從官網下載檔案，解壓上傳到最終目錄

因為本次試驗使用的是單機版的，因此直接在bin目錄下執行.start-cluster.sh即可，web埠為8081

至此，我們的環境已經全部準備好了

三、程式開發

3.1.程式生成日誌到flume

log4j配置：

log4j.rootLogger=INFO,flume,console
log4j.appender.console=org.apache.log4j.ConsoleAppender
log4j.appender.console.target=System.out
log4j.appender.console.layout=org.apache.log4j.PatternLayout
log4j.appender.console.layout.ConversionPattern="%d{yyyy-MM-dd HH:mm:ss} %p [%c:%L] - %m%n

log4j.appender.flume = org.apache.flume.clients.log4jappender.Log4jAppender
log4j.appender.flume.Hostname = 192.168.83.112
log4j.appender.flume.Port = 44444
log4j.appender.flume.UnsafeMode = true
log4j.appender.flume.layout=org.apache.log4j.PatternLayout
log4j.appender.flume.layout.ConversionPattern=%d{yyyy-MM-dd HH:mm:ss} %p [%c:%L] - %m%n

迴圈生成日誌

import java.util.Date;
import org.apache.log4j.Logger;
public class WriteLog {
    private static Logger logger = Logger.getLogger(WriteLog.class);
	public static void main(String[] args) throws InterruptedException {
	// 記錄debug級別的資訊  
        logger.debug("This is debug message.");  
        // 記錄info級別的資訊  
        logger.info("This is info message.");  
        // 記錄error級別的資訊  
        logger.error("This is error message.");
        int i = 0;
		while (true) {
			logger.info(new Date().getTime());
			logger.info("測試資料" + i);
			Thread.sleep(2000);
			i += 1;
		}
	}
}

在flume中的conf，複製模板為example.conf

書寫配置：

# Name the components on this agent
# 定義一個 agent 的元素

a1.sources = r1
a1.sinks = k1
a1.channels = c1

# Describe/configure the source
# 配置 source

#使用avro接收log4j過來的資料
a1.sources.r1.type = avro
a1.sources.r1.bind = 0.0.0.0
a1.sources.r1.port = 44444

# Describe the sink
# 配置 sink

#a1.sinks.k1.type = logger
#將資料寫入kafka，設定topic和brokers地址
a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink
a1.sinks.k1.topic = test
a1.sinks.k1.brokerList = 192.168.83.112:9092
a1.sinks.k1.requiredAcks = 1
a1.sinks.k1.batchSize = 100

# Use a channel which buffers events in memory
# 定義 channel

a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100

# Bind the source and sink to the channel
# 用 channel 連線起來 source 和 sink

a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

執行flume命令

flume-ng agent -c /opt/soft/apache-flume-1.8.0-bin/conf -f example.conf --name a1 -Dflume.root.logger=INFO,console

啟動java程式

3.2程式獲取kafka中的資料

java程式：

import org.apache.flume.*;
import org.apache.flume.conf.Configurable;
import org.apache.flume.sink.AbstractSink;

/**
 * Created by anan on 2018-7-31 14:20.
 */
public class CustomerSource extends AbstractSink implements Configurable {
    @Override
    public Status process() throws EventDeliveryException {
        Status status = null;
        // Start transaction
        Channel ch = getChannel();
        Transaction txn = ch.getTransaction();
        txn.begin();
        try {
            // This try clause includes whatever Channel operations you want to
            // do
            Event event = ch.take();
            // Send the Event to the external repository.
            // storeSomeData(e);
            String eventBody = new String(event.getBody(), "utf-8");

            System.out.println("============= " + eventBody + " ========");

            txn.commit();
            status = Status.READY;
        } catch (Throwable t) {
            txn.rollback();
            // Log exception, handle individual exceptions as needed
            status = Status.BACKOFF;

            // re-throw all Errors
            if (t instanceof Error) {
                throw (Error) t;
            }
        }
        // you must add this line of code in order to close the Transaction.
        txn.close();
        return status;
    }

    @Override
    public void configure(Context context) {

    }

    @Override
    public synchronized void start() {
        super.start();
    }

    @Override
    public synchronized void stop() {
        super.stop();
    }
}

新建flume conf test.conf

#soource的名字
agent.sources = kafkaSource
agent.channels = memoryChannel
agent.sinks = hdfsSink


agent.sources.kafkaSource.channels = memoryChannel
agent.sinks.hdfsSink.channel = memoryChannel

#-------- kafkaSource相關配置-----------------
agent.sources.kafkaSource.type = org.apache.flume.source.kafka.KafkaSource
agent.sources.kafkaSource.zookeeperConnect =192.168.83.112:2181
# 配置消費的kafka topic
agent.sources.kafkaSource.topic = test
# 配置消費者組的id
agent.sources.kafkaSource.groupId = flume
# 消費超時時間,參照如下寫法可以配置其他所有kafka的consumer選項。注意格式從kafka.xxx開始是consumer的配置屬性
agent.sources.kafkaSource.kafka.consumer.timeout.ms = 100



#------- memoryChannel相關配置-------------------------
agent.channels.memoryChannel.type = memory
agent.channels.memoryChannel.capacity=10000
agent.channels.memoryChannel.transactionCapacity=1000

#---------hdfsSink 相關配置------------------
agent.sinks.hdfsSink.type = com.gd.bigdataleran.flume.customerSource.CustomerSource

執行flume命令

flume-ng agent -c /opt/soft/apache-flume-1.8.0-bin/conf -f test.conf --name agent -Dflume.root.logger=INFO,console

然後檢視控制檯是否列印生成的日誌。

flink接收

如果想要flink接收kafka資料然後將資料經過簡單處理後放到kafka，就需要使用到flinkkafkaconsumer和flinkkafkaproductor

，java程式碼如下

package com.gd.bigdataleran.flink;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSink;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer09;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer09;
import org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicPartition;

import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;


/**
 * Created by anan on 2018-8-3 15:44.
 */
public class kafkaconsumer {
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000); // 非常關鍵，一定要設定啟動檢查點！！
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "192.168.83.112:9092");
//        properties.setProperty("zookeeper.connect", "192.168.83.112:2181");
        properties.setProperty("group.id", "test112");

        FlinkKafkaConsumer09<String> myConsumer = new FlinkKafkaConsumer09<>("test", new SimpleStringSchema(), properties);
        myConsumer.setStartFromEarliest();
        System.out.println("執行輸入");
        DataStream<String> stream = env.addSource(myConsumer);
        DataStream ds = stream.map(new MapFunction<String, Object>() {
            @Override
            public Object map(String s) throws Exception {
                return s + "==" + new Date().getTime();
            }
        });

        FlinkKafkaProducer09<String> flinkKafkaProducer09 = new FlinkKafkaProducer09<String>("192.168.83.112:9092","test1",new SimpleStringSchema());
        ds.addSink(flinkKafkaProducer09);
        System.out.println("執行輸出");

        env.execute();
    }
}

將需要的jar包上傳到flink目錄下的lib目錄下，然後執行flink命令即可

 flume-ng agent -c /opt/soft/apache-flume-1.8.0-bin/conf -f example.conf --name a1 -Dflume.root.logger=INFO,console

驗證資料

使用java連線kafka，獲取topic中的資料進行驗證；驗證程式碼 如下

    private void getKafkaData() {
        String topic = "test1";
        Properties kafkaProps = new Properties();
        kafkaProps.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        kafkaProps.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        kafkaProps.put("bootstrap.servers", "192.168.83.112:9092");
        kafkaProps.put("zookeeper.connect", "192.168.83.112:2181");
        kafkaProps.put("group.id", "farmtest1");
        kafkaProps.put("auto.offset.reset", "smallest");
        kafkaProps.put("enable.auto.commit", "true");
        ConsumerConnector consumer = Consumer.createJavaConsumerConnector(new ConsumerConfig(kafkaProps));

        Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
        topicCountMap.put(topic, 1); // 一次從主題中獲取一個數據
        Map<String, List<KafkaStream<byte[], byte[]>>> messageStreams = consumer.createMessageStreams(topicCountMap);
        KafkaStream<byte[], byte[]> stream = messageStreams.get(topic).get(0);// 獲取每次接收到的這個資料
        ConsumerIterator<byte[], byte[]> iterator = stream.iterator();
        while (iterator.hasNext()) {
            String message = new String(iterator.next().message());
            consumer.commitOffsets();
            System.out.println(message);
        }
    }

時間差匆忙，有問題歡迎大家提問討論