kafka-Storm中如何將日志文件打印到local

這篇文章給大家介紹kafka-Storm中如何將日志文件打印到local，內(nèi)容非常詳細(xì)，感興趣的小伙伴們可以參考借鑒，希望對(duì)大家能有所幫助。

閱讀前提：

        1 ： 您可能需要對(duì)  logback 日志系統(tǒng)有所了解

        2 ：您可能需要對(duì)于 kafka 有初步的了解

        3：請(qǐng)代碼查看之前，請(qǐng)您仔細(xì)參考系統(tǒng)的業(yè)務(wù)圖解

    由于kafka本身自帶了和『Hadoop』的接口，如果需要將kafka中的文件直接遷移到HDFS，請(qǐng)參看本ID的另外一篇博文：

        業(yè)務(wù)系統(tǒng)-kafka-Storm【日志本地化】 - 2 ：直接通過(guò)kafka將日志傳遞到HDFS

    1： 一個(gè)正式環(huán)境系統(tǒng)的系統(tǒng)設(shè)計(jì)圖解：

              通過(guò)kafka集群，在2個(gè)相同的topic之下，通過(guò)kafka-storm, he kafka-hadoop，2 個(gè)Consumer，針對(duì)同樣的一份數(shù)據(jù)，我們分流了2個(gè)管道：

            其一： 實(shí)時(shí)通道

            其二：離線通道

       在日志本地化的過(guò)程之中，前期，由于日志的清洗，過(guò)濾的工作是放在Storm集群之中，也就是說(shuō)，留存到本地locla的日志。是我們?cè)赟torm集群之中進(jìn)行了清洗的數(shù)據(jù)。

      也就是：

            如下圖所示：

      在kafka之中，通常而言，有如下的 代碼 用來(lái)處理：

         在這里我們針對(duì)了2種日志，有兩個(gè)Consumer用來(lái)處理

package com.mixbox.kafka.consumer;

public class logSave {

	public static void main(String[] args) throws Exception {

		Consumer_Thread visitlog = new Consumer_Thread(KafkaProperties.visit);
		visitlog.start();

		Consumer_Thread orderlog = new Consumer_Thread(KafkaProperties.order);
		orderlog.start();

	}
}

     在這里，我們依據(jù)不同的原始字段，將不同的數(shù)據(jù)保存到不同的文件之中。

package com.mixbox.kafka.consumer;

import java.io.UnsupportedEncodingException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;
import kafka.message.MessageAndMetadata;

/**
 * @author Yin Shuai
 */
public class Consumer_Thread extends Thread {

	// 在事實(shí)上我們會(huì)依據(jù)傳遞的topic名稱，來(lái)生成不桐的記錄機(jī)器
	// private Logger _log_order = LoggerFactory.getLogger("order");
	// private Logger _log_visit = LoggerFactory.getLogger("visit");

	private Logger _log = null;

	private final ConsumerConnector _consumer;
	private final String _topic;

	public Consumer_Thread(String topic) {

		_consumer = kafka.consumer.Consumer
				.createJavaConsumerConnector(createConsumerConfig());
		this._topic = topic;

		_log = LoggerFactory.getLogger(_topic);

		System.err.println("log的名稱" + _topic);

	}

	private static ConsumerConfig createConsumerConfig() {
		Properties props = new Properties();
		props.put("zookeeper.connect", KafkaProperties.zkConnect);
		// 在這里我們的組ID為logSave
		props.put("group.id", KafkaProperties.logSave);
		props.put("zookeeper.session.timeout.ms", "100000");
		props.put("zookeeper.sync.time.ms", "200");
		props.put("auto.commit.interval.ms", "1000");
		return new ConsumerConfig(props);

	}

	public void run() {

		Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
		topicCountMap.put(_topic, new Integer(1));

		Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap = _consumer
				.createMessageStreams(topicCountMap);

		for (KafkaStream<byte[], byte[]> kafkaStream : consumerMap.get(_topic)) {
			ConsumerIterator<byte[], byte[]> iterator = kafkaStream.iterator();
			while (iterator.hasNext()) {
				MessageAndMetadata<byte[], byte[]> next = iterator.next();
				try {

					// 在這里我們分拆了一個(gè)Consumer 來(lái)處理visit日志
					logFile(next);
					System.out.println("message:"
							+ new String(next.message(), "utf-8"));
				} catch (UnsupportedEncodingException e) {
					e.printStackTrace();
				}
			}
		}
	}

	private void logFile(MessageAndMetadata<byte[], byte[]> next)
			throws UnsupportedEncodingException {
		_log.info(new String(next.message(), "utf-8"));
	}

}

    一個(gè)簡(jiǎn)單的小tips：

        logback.xml  ,提醒您注意，這里的配置文件太過(guò)粗淺。如有需要，請(qǐng)自行填充。

<?xml version="1.0" encoding="UTF-8" ?>
<configuration>

	<jmxConfigurator />
	<!-- 控制臺(tái)輸出日志 -->
	<appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">

		<!-- 過(guò)濾掉 TRACE 和 DEBUG 級(jí)別的日志 -->
		<!-- <filter class="ch.qos.logback.classic.filter.ThresholdFilter"> -->
		<!-- <level>INFO</level> -->
		<!-- </filter> -->

		<!-- 按天來(lái)回滾，如果需要按小時(shí)來(lái)回滾，則設(shè)置為{yyyy-MM-dd_HH} -->
		<rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
			<fileNamePattern>f:/opt/log/test.%d{yyyy-MM-dd}.log</fileNamePattern>
			<!-- 如果按天來(lái)回滾，則最大保存時(shí)間為1天，1天之前的都將被清理掉 -->
		</rollingPolicy>

		<!-- 日志輸出格式 -->
		<layout class="ch.qos.logback.classic.PatternLayout">
			<pattern>
				%d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level
				%logger{36}-%msg%n</pattern>
		</layout>
	</appender>



	<!-- 記錄到日志 文件的滾動(dòng)日志 -->
	<appender name="ERROR"
		class="ch.qos.logback.core.rolling.RollingFileAppender">

		<file>
			e:/logs/error/error.log
		</file>
		<filter class="ch.qos.logback.classic.filter.LevelFilter">
			<level>
				ERROR
			</level>
			<onMatch>ACCEPT</onMatch>
			<onMismatch>DENY</onMismatch>
		</filter>
		<!-- 定義每天生成一個(gè)日志文件 -->
		<rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
			<fileNamePattern>e:/logs/yuanshi-%d{yyyy-MM-dd}.log</fileNamePattern>
			<MaxHistory>10</MaxHistory>
		</rollingPolicy>

		<!-- 日志樣式 -->
		<layout class="ch.qos.logback.classic.PatternLayout">
			<pattern>%d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level
				%logger{36}-%msg%n</pattern>
		</layout>
	</appender>


	<!-- 記錄到日志 文件的滾動(dòng)日志 -->
	<appender name="FILE"
		class="ch.qos.logback.core.rolling.RollingFileAppender">

		<file>E:\logs\file\file.log</file>

		<filter class="ch.qos.logback.classic.filter.LevelFilter">
			<level>INFO</level>
			<onMatch>ACCEPT</onMatch>
			<onMismatch>DENY</onMismatch>
		</filter>

		<!-- 定義每天生成一個(gè)日志文件 -->
		<rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
			<fileNamePattern>e:/logs/venality-%d{yyyy-MM-dd}.log
			</fileNamePattern>
			<MaxHistory>10</MaxHistory>
		</rollingPolicy>

		<!-- 日志樣式 -->
		<layout class="ch.qos.logback.classic.PatternLayout">
			<pattern>%d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level
				%logger{36}-%msg%n</pattern>
		</layout>
	</appender>


	<appender name="visit"
	class="ch.qos.logback.core.rolling.RollingFileAppender">
		<File>
			E:\logs\visitlog\visit.log
		</File>
		<encoder>
			<pattern>%msg%n</pattern>
		</encoder>

		<filter class="ch.qos.logback.classic.filter.ThresholdFilter">
			<level>INFO</level>
		</filter>
		<rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
			<fileNamePattern>E:\logs\visit.log.%d{yyyy-MM-dd}
			</fileNamePattern>
		</rollingPolicy>
	</appender>
	<logger name="visit" additivity="false" level="INFO">
		<appender-ref ref="visit" />
	</logger>


	<appender name="order"
		class="ch.qos.logback.core.rolling.RollingFileAppender">
		<File>
			E:\logs\orderlog\order.log
		</File>
		<encoder>
			<pattern>%msg%n
			</pattern>
		</encoder>

		<filter class="ch.qos.logback.classic.filter.ThresholdFilter">
			<level>INFO</level>
		</filter>
		<rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
			<fileNamePattern>E:\logs\order.log.%d{yyyy-MM-dd}
			</fileNamePattern>
		</rollingPolicy>
	</appender>
	<logger name="order" additivity="false" level="INFO">
		<appender-ref ref="order" />
	</logger>


	<root level="DEBUG">
		<appender-ref ref="FILE" />
	</root>
</configuration>

關(guān)于kafka-Storm中如何將日志文件打印到local就分享到這里了，希望以上內(nèi)容可以對(duì)大家有一定的幫助，可以學(xué)到更多知識(shí)。如果覺(jué)得文章不錯(cuò)，可以把它分享出去讓更多的人看到。