Flume installation and use

Flume Installation and use of

Flume Installation

1、 Upload to virtual machine , And extract the

tar -zxvf apache-flume-1.9.0-bin.tar.gz -C /usr/local/soft/

Add the following command to the environment variable , have access to soft Quickly switch to /usr/local/soft

alias soft=‘cd /usr/local/soft/’

2、 Rename Directory , And configure environment variables

mv apache-flume-1.9.0-bin/ flume-1.9.0
vim /etc/profile
source /etc/profile

3、 see flume edition

flume-ng version

[[email protected] soft]# flume-ng version
Flume 1.9.0
Source code repository: https://git-wip-us.apache.org/repos/asf/flume.git
Revision: d4fcab4f501d41597bc616921329a4339f73585e
Compiled by fszabo on Mon Dec 17 20:45:25 CET 2018
From source with checksum 35db629a3bda49d23e9b3690c80737f9
[[email protected] soft]# 

4、 test flume

• Monitor a directory , Print out the data

  • The configuration file

  #  First, first agent Give a name   It's called a1
  #  Give to separately source channel sink Name 
  a1.sources = r1
  a1.channels = c1
  a1.sinks = k1
  
  #  Respectively for source、channel、sink To configure 
  
  #  To configure source
  #  take source The type of is specified as  spooldir  Used to listen for changes in files in a directory 
  #  Because each component may have the same attribute name , So when configuring each component  
  #  Need to add  agent Name .sources. The name of the component . attribute  =  Property value 
  a1.sources.r1.type = spooldir
  a1.sources.r1.spoolDir = /root/data/
  a1.sources.r1.fileSuffix = .ok
  a1.sources.r1.fileHeader = true
  
  #  to r1 This souces Configure an interceptor and name it  i1
  a1.sources.r1.interceptors = i1
  #  Will intercept i1 The type of is set to timestamp  The data processing time is inserted in milliseconds event Of header in 
  # a1.sources.r1.interceptors.i1.type = timestamp
  #  Will intercept i1 The type of is set to regex_filter  Will filter data based on regular expressions 
  a1.sources.r1.interceptors.i1.type = regex_filter
  #  Configure regular expressions 
  a1.sources.r1.interceptors.i1.regex = \\d{3,6}
  # excludeEvents = true  Indicates the filter to be matched , Unmatched release 
  a1.sources.r1.interceptors.i1.excludeEvents = true
  
  #  To configure sink
  #  Use logger As sink Components , The collected data can be printed directly to the console 
  a1.sinks.k1.type = logger
  
  #  To configure channel
  #  take channel The type of is set to memory, It means that you will event The cache is stored in memory 
  a1.channels.c1.type = memory
  
  #  assemble 
  #  take sources Of channels Property specified as c1
  a1.sources.r1.channels = c1
  
  #  take sinks Of channel Property specified as c1
  a1.sinks.k1.channel = c1
  

  • start-up agent

  flume-ng agent -n a1 -f ./spoolingtest.conf -Dflume.root.logger=DEBUG,console
  

  • newly build /root/data Catalog

  mkdir /root/data
  

  • stay /root/data/ New file in directory , Input content , Observe flume Log printed by the process

  #  At will a.txt Add some content to 
  vim /root/data/a.txt
  

5、flume Use

• spoolingToHDFS.conf

  • The configuration file

  # a Said to agent Name it a
  #  to source The component is named r1
  a.sources = r1
  #  to sink The component is named k1
  a.sinks = k1 
  #  to channel The component is named c1
  a.channels = c1
  # Appoint spooldir Properties of 
  a.sources.r1.type = spooldir 
  a.sources.r1.spoolDir = /root/data 
  a.sources.r1.fileHeader = true 
  a.sources.r1.interceptors = i1 
  a.sources.r1.interceptors.i1.type = timestamp
  # Appoint sink The type of 
  a.sinks.k1.type = hdfs
  a.sinks.k1.hdfs.path = /flume/data/dir1
  #  Specify the filename prefix 
  a.sinks.k1.hdfs.filePrefix = student
  #  Specify how much data to write to the file once   Company ：bytes
  a.sinks.k1.hdfs.rollSize = 102400
  #  Specify how many files to write once 
  a.sinks.k1.hdfs.rollCount = 1000
  #  Specifies that the file type is   flow   What comes out 
  a.sinks.k1.hdfs.fileType = DataStream
  #  Specifies the file output format   by text
  a.sinks.k1.hdfs.writeFormat = text
  #  Specify the file name suffix 
  a.sinks.k1.hdfs.fileSuffix = .txt
  
  # Appoint channel
  a.channels.c1.type = memory 
  a.channels.c1.capacity = 1000
  #  Express sink Every time from channel How much data to take from the 
  a.channels.c1.transactionCapacity = 100
  #  assemble 
  a.sources.r1.channels = c1 
  a.sinks.k1.channel = c1
  

  • stay /root/data/ Prepare data under the directory

  The Zen of Python, by Tim Peters
  
  Beautiful is better than ugly.
  Explicit is better than implicit.
  Simple is better than complex.
  Complex is better than complicated.
  Flat is better than nested.
  Sparse is better than dense.
  Readability counts.
  Special cases aren't special enough to break the rules.
  Although practicality beats purity.
  Errors should never pass silently.
  Unless explicitly silenced.
  In the face of ambiguity, refuse the temptation to guess.
  There should be one-- and preferably only one --obvious way to do it.
  Although that way may not be obvious at first unless you're Dutch.
  Now is better than never.
  Although never is often better than *right* now.
  If the implementation is hard to explain, it's a bad idea.
  If the implementation is easy to explain, it may be a good idea.
  Namespaces are one honking great idea -- let's do more of those!
  

  • start-up agent

  flume-ng agent -n a -f ./spoolingToHDFS.conf -Dflume.root.logger=DEBUG,console
  

• hbaseLogToHDFS

  • The configuration file

  # a Said to agent Name it a
  #  to source The component is named r1
  a.sources = r1
  #  to sink The component is named k1
  a.sinks = k1 
  #  to channel The component is named c1
  a.channels = c1
  # Appoint exec Properties of 
  a.sources.r1.type = exec 
  a.sources.r1.command = tail -f /usr/local/soft/hbase-1.4.6/logs/hbase-root-master-master.log
  # Appoint sink The type of 
  a.sinks.k1.type = hdfs
  a.sinks.k1.hdfs.path = /flume/data/dir2
  #  Specify the filename prefix 
  a.sinks.k1.hdfs.filePrefix = hbaselog
  #  Specify how much data to write to the file once   Company ：bytes
  a.sinks.k1.hdfs.rollSize = 102400
  #  Specify how many files to write once 
  a.sinks.k1.hdfs.rollCount = 1000
  #  Specifies that the file type is   flow   What comes out 
  a.sinks.k1.hdfs.fileType = DataStream
  #  Specifies the file output format   by text
  a.sinks.k1.hdfs.writeFormat = text
  #  Specify the file name suffix 
  a.sinks.k1.hdfs.fileSuffix = .txt
  
  # Appoint channel
  a.channels.c1.type = memory 
  a.channels.c1.capacity = 1000
  #  Express sink Every time from channel How much data to take from the 
  a.channels.c1.transactionCapacity = 100
  #  assemble 
  a.sources.r1.channels = c1 
  a.sinks.k1.channel = c1
  

• hbaselogToHBase

  • stay hbase Created in log surface

  create 'log','cf1'
  

  • The configuration file

  # a Said to agent Name it a
  #  to source The component is named r1
  a.sources = r1
  #  to sink The component is named k1
  a.sinks = k1 
  #  to channel The component is named c1
  a.channels = c1
  # Appoint exec Properties of 
  a.sources.r1.type = exec 
  a.sources.r1.command = cat /usr/local/soft/hbase-1.4.6/logs/hbase-root-master-master.log
  # Appoint sink The type of 
  a.sinks.k1.type = hbase
  a.sinks.k1.table = log
  a.sinks.k1.columnFamily = cf1
  
  # Appoint channel
  a.channels.c1.type = memory 
  a.channels.c1.capacity = 100000
  #  Express sink Every time from channel How much data to take from the 
  a.channels.c1.transactionCapacity = 100
  #  assemble 
  a.sources.r1.channels = c1 
  a.sinks.k1.channel = c1
  

• netcatLogger

  monitor telnet port

  • install telnet

  yum install telnet
  

  • The configuration file

  # a Said to agent Name it a
  #  to source The component is named r1
  a.sources = r1
  #  to sink The component is named k1
  a.sinks = k1 
  #  to channel The component is named c1
  a.channels = c1
  # Appoint netcat Properties of 
  a.sources.r1.type = netcat 
  a.sources.r1.bind = 0.0.0.0 
  a.sources.r1.port = 8888 
  
  # Appoint sink The type of 
  a.sinks.k1.type = logger
  # Appoint channel
  a.channels.c1.type = memory 
  a.channels.c1.capacity = 1000
  #  Express sink Every time from channel How much data to take from the 
  a.channels.c1.transactionCapacity = 100
  #  assemble 
  a.sources.r1.channels = c1 
  a.sinks.k1.channel = c1
  

  • start-up

    • Start... First agent

    flume-ng agent -n a -f ./netcatToLogger.conf -Dflume.root.logger=DEBUG,console
    

    • Start up telnet

    telnet master 8888
    

• httpToLogger

  • The configuration file

  # a Said to agent Name it a
  #  to source The component is named r1
  a.sources = r1
  #  to sink The component is named k1
  a.sinks = k1 
  #  to channel The component is named c1
  a.channels = c1
  # Appoint http Properties of 
  a.sources.r1.type = http
  a.sources.r1.port = 6666 
  
  # Appoint sink The type of 
  a.sinks.k1.type = logger
  # Appoint channel
  a.channels.c1.type = memory 
  a.channels.c1.capacity = 1000
  #  Express sink Every time from channel How much data to take from the 
  a.channels.c1.transactionCapacity = 100
  #  assemble 
  a.sources.r1.channels = c1 
  a.sinks.k1.channel = c1
  

  • start-up

    • Start... First agent

    flume-ng agent -n a -f ./httpToLogger.conf -Dflume.root.logger=DEBUG,console
    

    • Reuse curl To launch a http request

    curl -X POST -d '[{ "headers" :{"a" : "a1","b" : "b1"},"body" : "hello~http~flume~"},{ "headers" :{"a2" : "a11","b2" : "b11"},"body" : "hello~http~flume2~"},{ "headers" :{"a3" : "a22","b3" : "b22"},"body" : "hello~http~flume3~"}]' http://master:6666