zoukankan      html  css  js  c++  java
  • flume 以 kafka 为channel 的配置

    #此配置以kafka的一个topic为channel,相比其他channel类型 file和cache 兼并了快和安全的要求!


    # Define a kafka channel

    a1.channels.c1.type = org.apache.flume.channel.kafka.KafkaChannel
    a1.channels.c1.kafka.bootstrap.servers = kafka-1:9092,kafka-2:9092,kafka-3:9092
    a1.channels.c1.kafka.topic = channel1
    a1.channels.c1.kafka.consumer.group.id = flume-consumer
    # Define an spooldir source
    a1
    .sources.src.type = spooldir
    a1
    .sources.src.spoolDir = /home/flumeSpool
    a1
    .sources.src.deletePolicy=immediate
    #为了把数据分发到kafka topic的所有partition

    a1.sources.src.interceptors = i2
    a1.sources.src.interceptors.i2.type=org.apache.flume.sink.solr.morphline.UUIDInterceptor$Builder 
    a1
    .sources.src.interceptors.i2.headerName=key
    a1
    .sources.src.interceptors.i2.preserveExisting=false

    # Define a kafka sink

    a1
    .sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink
    a1
    .sinks.k1.topic = web_nginx_log
    a1
    .sinks.k1.brokerList =
    kafka-1:9092,kafka-2:9092,kafka-3:9092
    
    
    a1.sinks.k1.kafka.producer.acks = 1
    a1.sinks.k1.kafka.producer.linger.ms = 1
    a1.sinks.ki.kafka.producer.compression.type = snappy
    
    

    # Finally, now that we've defined all of our components, tell
    a1
    .channels = c1
    a1
    .sources = src
    a1
    .sinks = k1

    # Bind the source and sink to the channel
    a1
    .sources.src.channels = c1
    a1
    .sinks.k1.channel = c1
  • 相关阅读:
    oracle数据库常用指令
    MySql常用命令
    js动态添加和删除行
    Mybatis模糊查询
    laravel 成功跳转页面
    laravel 验证码
    git获取公钥和私钥以及常用的命令
    laravel css和js的引入
    git add -A 、git add -u 、 git add . 三种区别
    windows下github 出现Permission denied (publickey).解决方法
  • 原文地址:https://www.cnblogs.com/RichardYD/p/6283748.html
Copyright © 2011-2022 走看看