Flink Stream日志写入Kafka集群

向右看齐 2021-12-12 15:51 357阅读 0赞

## Flink Stream日志写入Kafka集群 ##

Flink Stream的输出日志，直接输出的Kafka集群，进行持久化。

#### 一、log4j.properties文件配置 ####

log4j.rootLogger=INFO,file,kafka
    
    log4j.logger.org.apache.kafka=INFO
    
    ###################################################
    # 输出的日志，写入文件
    ###################################################
    # appender file
    log4j.appender.file=org.apache.log4j.RollingFileAppender
    #设置log文件名
    log4j.appender.file.file=logs/test.log
    #log是否追加
    log4j.appender.file.append=true
    log4j.appender.file.layout=org.apache.log4j.PatternLayout
    log4j.appender.file.layout.ConversionPattern=%d{yyyy-MM-dd HH:mm:ss,SSS} %-5p %-60c %x - %m%n
    #文件大小
    log4j.appender.file.MaxFileSize=100MB
    #文件保留个数
    log4j.appender.file.MaxBackupIndex=5
    
    #####################################################
    # 输出的日志写入kafka
    #####################################################
    #输出日志到kafka
    log4j.appender.kafka=org.apache.kafka.log4jappender.KafkaLog4jAppender
    #设置topic
    log4j.appender.kafka.topic=pro-log
    #设置kafka连接地址
    log4j.appender.kafka.brokerList=localhost:9092
    log4j.appender.kafka.compressionType=none
    log4j.appender.kafka.requiredNumAcks=0
    #设置是否异步
    log4j.appender.kafka.syncSend=false
    log4j.appender.kafka.layout=org.apache.log4j.PatternLayout
    log4j.appender.kafka.layout.ConversionPattern=%d [%-5p] [%t] - [%l] %m%n
    # Suppress the irrelevant (wrong) warnings from the Netty channel handler
    log4j.logger.org.apache.flink.shaded.akka.org.jboss.netty.channel.DefaultChannelPipeline=ERROR, file

#### 二、测试代码 ####

public class KafkaDemo { 
        public static void main(String[] args) throws Exception { 
    
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
            env.setParallelism(1);
    
            Properties p = new Properties();
            p.setProperty("bootstrap.servers", "localhost:9092");
    
            DataStreamSource<String> ds = env.addSource(new FlinkKafkaConsumer010<String>("pro-log2", new SimpleStringSchema(), p));
            ds.print();
    
            env.execute("KafkaDemo");
        }
    }

#### 三、消费Kafka的数据 ####

./kafka-console-consumer.sh --zookeeper localhost:2181 --from-beginning --topic pro-log