FlinkStream的输出日志，直接输出的Kafka集群，进行持久化。
一、log4j.properties文件配置log4j.rootLogger=INFO,file,kafkalog4j.logger.org.apache.kafka=INFO####################################################输出的日志，写入文件####################################################appenderfilelog4j.appender.file=org.apache.log4j.RollingFileAppender#设置log文件名log4j.appender.file.file=logs/test.log#log是否追加log4j.appender.file.append=truelog4j.appender.file.layout=org.apache.log4j.PatternLayoutlog4j.appender.file.layout.ConversionPattern=%d{yyyy-MM-ddHH:mm:ss,SSS}%-5p%-60c%x-%m%n#文件大小log4j.appender.file.MaxFileSize=100MB#文件保留个数log4j.appender.file.MaxBackupIndex=5######################################################输出的日志写入kafka######################################################输出日志到kafkalog4j.appender.kafka=org.apache.kafka.log4jappender.KafkaLog4jAppender#设置topiclog4j.appender.kafka.topic=pro-log#设置kafka连接地址log4j.appender.kafka.brokerList=localhost:9092log4j.appender.kafka.compressionType=nonelog4j.appender.kafka.requiredNumAcks=0#设置是否异步log4j.appender.kafka.syncSend=falselog4j.appender.kafka.layout=org.apache.log4j.PatternLayoutlog4j.appender.kafka.layout.ConversionPattern=%d[%-5p][%t]-[%l]%m%n#Suppresstheirrelevant(wrong)warningsfromtheNettychannelhandlerlog4j.logger.org.apache.flink.shaded.akka.org.jboss.netty.channel.DefaultChannelPipeline=ERROR,file二、测试代码publicclassKafkaDemo{publicstaticvoidmain(String[]args)throwsException{StreamExecutionEnvironmentenv=StreamExecutionEnvironment.getExecutionEnvironment();env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);env.setParallelism(1);Propertiesp=newProperties();p.setProperty("bootstrap.servers","localhost:9092");DataStreamSource<String>ds=env.addSource(newFlinkKafkaConsumer010<String>("pro-log2",newSimpleStringSchema(),p));ds.print();env.execute("KafkaDemo");}}三、消费Kafka的数据./kafka-console-consumer.sh--zookeeperlocalhost:2181--from-beginning--topicpro-log
