1. agent.channels.ch1.type = memory  
    2. agent.sources.sql-source.channels = ch1  
    3. agent.channels = ch1  
    4. agent.sinks = HDFS  
    6. agent.sources = sql-source  
    7. agent.sources.sql-source.type = org.keedio.flume.source.SQLSource  
    9. agent.sources.sql-source.connection.url = jdbc:mysql://172.16.1.127:3306/test  
    10. agent.sources.sql-source.user = root  
    11. agent.sources.sql-source.password = 123456  
    12. agent.sources.sql-source.table = wlslog  
    13. agent.sources.sql-source.columns.to.select = *  
    15. agent.sources.sql-source.incremental.column.name = id  
    16. agent.sources.sql-source.incremental.value = 0  
    18. agent.sources.sql-source.run.query.delay=5000  
    20. agent.sources.sql-source.status.file.path = /var/lib/flume  
    21. agent.sources.sql-source.status.file.name = sql-source.status  
    23. agent.sinks.HDFS.type = logger
    25. -----  
    26. agent.sinks.HDFS.channel = ch1  
    27. agent.sinks.HDFS.type = hdfs  
    28. agent.sinks.HDFS.hdfs.path = hdfs://mycluster/flume/mysql  
    29. agent.sinks.HDFS.hdfs.fileType = DataStream  
    30. agent.sinks.HDFS.hdfs.writeFormat = Text  
    31. agent.sinks.HDFS.hdfs.rollSize = 268435456  
    32. agent.sinks.HDFS.hdfs.rollInterval = 0  
    33. agent.sinks.HDFS.hdfs.rollCount = 0  
    34. ------
    1. f2.sources = r1
    2. f2.channels = c1
    3. f2.sinks = k1
    5. # 这里用 自己定义的 SQLSource
    6. f2.sources.r1.type = org.keedio.flume.source.SQLSource
    7. f2.sources.r1.connectionurl = jdbc:mysql://172.16.1.127:3306/test 
    8. f2.sources.r1.user = yyj
    9. f2.sources.r1.password = yyj
    10. #f2.sources.r1.driverclass= oracle.jdbc.driver.OracleDriver
    11. f2.sources.r1.filepath = /var/log/sqllog
    12. f2.sources.r1.filename = sqlSource.status
    13. #f2.sources.r1.customquery = select a.id,a.COUPON_id,b.id from USER_COUPON_CODE_1 a ,COUPON_CODE b  where a.COUPON_id = b.id
    14. #f2.sources.r1.customquery = select a.id,a.COUPON_id,b.id from USER_COUPON_CODE_1 a ,COUPON_CODE b  where a.COUPON_id = b.id  and  a.id > $@$
    15. f2.sources.r1.customquery = select * from USER_COUPON_CODE_1
    16. f2.sources.r1.begin = 0
    17. f2.sources.r1.autoincrementfield = a.id
    18. f2.sources.r1.batchsize = 1000
    19. #具体定义channel
    20. f2.channels.c1.type = memory
    21. f2.channels.c1.capacity = 1000
    22. f2.channels.c1.transactionCapacity = 100
    23. #具体定义sink
    24. f2.sinks.k1.type = logger
    25. #组装source、channel、sink
    26. f2.sources.r1.channels = c1
    27. f2.sinks.k1.channel = c1

    测试过:

    1. agent.sources = mysql
    2. agent.sinks = k1
    3. agent.channels = ch1 
    6. agent.sources.mysql.type = org.keedio.flume.source.SQLSource
    7. agent.sources.mysql.hibernate.connection.url = jdbc:mysql://192.168.1.64:3306/test
    9. agent.sources.mysql.hibernate.connection.user = root
    10. agent.sources.mysql.hibernate.connection.password = 123456
    11. agent.sources.mysql.table = wlslog 
    12. agent.sources.mysql.columns.to.select = *  
    13. agent.sources.mysql.hibernate.connection.autocommit = true
    15. agent.sources.mysql.incremental.column.name = id  
    16. agent.sources.mysql.incremental.value = 0 
    18. agent.sources.mysql.status.file.path = /var/lib/flume  
    19. agent.sources.mysql.status.file.name = sql-source.status1
    21. agent.channels.ch1.type = memory
    22. agent.channels.ch1.capacity = 10000
    23. agent.channels.ch1.transactionCapacity = 1000
    25. agent.sinks.k1.type=logger
    26. agent.sources.mysql.channels=ch1
    27. agent.sinks.k1.channel=ch1