Flume learning notes

brief introduction

summary

Basic concepts

Flow model

1. Single stage flow

2. Multistage flow
   

3. Fan in flow
   

4. Fan out flow
   

introduction

3. Write format file
stay data Create a file in the directory , file name , The file suffix is optional
# to agent,source Naming
a1.sources=s1
# to channel Naming
a1.channels=c1
# to sink Naming
a1.sinks=k1

# To configure Source
a1.sources.s1.type=netcat // Means monitor tcp request , hold tcp The requested content is used as log data
a1.sources.s1.bind=0.0.0.0 // Configure the listening node ip
a1.sources.s1.port=8090 // Configure listening port
# To configure Channel
a1.channels.c1.type=memory // Means to temporarily store data in memory
a1.channels.c1.capacity=10000 // It means that tenthousand pieces of data can be temporarily stored in memory
a1.channels.c1.transactionCapacity=100 // Indicates the transaction capacity
# To configure sink
a1.sinks.k1.type=logger // Print data to the console

# take Source and Channel Binding
a1.sources.s1.channels=c1
# take Sink and Channel Binding
a1.sinks.k1.channel=c1

Execute startup command ：
…/bin/flume-ng agent -n a1 -c …/conf -f basic.conf -Dflume.root.logger=INFO,console

New window , Carry out orders
nc hadoop01 8090 //nc To express to hadoop01 Of 8090 Port send a tcp request

Components

One .Source

1.AVRO Source

# Edit format file , Go to this directory first
/home/software/apache-flume-1.7.0-bin/data
vim avrosource.conf
a1.sources=s1
a1.channels=c1
a1.sinks=k1
a1.sources.s1.type=avro
a1.sources.s1.bind=0.0.0.0
a1.sources.s1.port=8090
a1.channels.c1.type=memory
a1.sinks.k1.type=logger
a1.sources.s1.channels=c1
a1.sinks.k1.channel=c1
# Specify start command
…/bin/flume-ng agent -n a1 -c …/conf -f avrosource.conf -Dflume.root.logger=INFO,console
# send out AVRO Serialized data
# Open a new window
cd /home/software/apache-flume-1.7.0-bin/data
vim avrotest # Add content to this file
# Serialize the file and send
…/bin/flume-ng avro-client -H hadoop01 -p 8090 -c …/conf -F avrotest

Exec Source

Spooling Directory Source

HTTP Source

Customize source

1. demand ： simulation ：Sequence Generator Source

package cn.tedu.flume.source;

import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.EventDrivenSource;
import org.apache.flume.channel.ChannelProcessor;
import org.apache.flume.conf.Configurable;
import org.apache.flume.event.EventBuilder;
import org.apache.flume.source.AbstractSource;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

//  simulation ：Sequence Generator Source
public class AuthSource extends AbstractSource
        implements EventDrivenSource, Configurable {
    

    private ExecutorService es;
    private int step;

    //  Get the value of the specified property 
    @Override
    public void configure(Context context) {
    
        //  If you specify step Value , Then it will be incremented according to the specified step 
        //  If not specified step, Then the default auto increment step is 1
        // This is actually the configuration file a1.sources.s1.step The attribute value ,
        // Subsequently, the string contents of the first parameter specified are different , You can get different attribute values , The back 1
        // Express , If you do not specify a default value 
        step = context.getInteger("step", 1);
    }

    @Override
    public synchronized void start() {
    
        es = Executors.newFixedThreadPool(5);
        //  obtain Channel
        ChannelProcessor cp = this.getChannelProcessor();
        es.submit(new Add(step, cp));
    }

    @Override
    public synchronized void stop() {
    
        if (es != null) {
    
            es.shutdown();
        }
    }

}

class Add implements Runnable {
    

    private final int step;
    private final ChannelProcessor cp;

    public Add(int step, ChannelProcessor cp) {
    
        this.step = step;
        this.cp = cp;
    }

    @Override
    public void run() {
    
        int i = 0;
        while (true) {
    
            //  encapsulation headers
            Map<String, String> headers = new HashMap<>();
            headers.put("date", System.currentTimeMillis() + "");
            //  encapsulation body
            byte[] body = (i + "").getBytes();
            //  stay Flume in , Each log collected will be encapsulated into a Event object 
            Event e = EventBuilder.withBody(body, headers);
            //  take Event Objects in the Channel in 
            cp.processEvent(e);
            i += step;
        }
    }

}

Package the project , take jar Put the bag in /home/software/apache-flume-1.7.0-bin/lib Under the table of contents
Then format the configuration file auth.conf
a1.sources=s1

a1.channels=c1

a1.sinks=k1
a1.sources.s1.type=cn.tedu.flume.source.AuthSource # Note that it should be written as the full path name of the class
a1.sources.s1.port=8090

a1.channels.c1.type=memory
a1.sinks.k1.type=logger
a1.sources.s1.channels=c1
a1.sinks.k1.channel=c1
Then use this configuration file to start the customization source that will do

Channel

1. Memory Channel
   
2. File Channel
   
   

Sink

1. HDFS Sink
   
   
2. Logger Sink
   
3. File Roll Sink
   
4. AVRO Sink( Realize multi-level flow 、 Fan in flow and fan out flow ）
   
   Multistage flow
   
   Fan in flow
   
   Delete flow
   

Customize sink

Custom Sink: You need to customize a class implementation Sink Interface , In the process of implementation , Matters needing attention .
Case study ： Write out the data to the specified path

package cn.tedu.flume.sink;

import org.apache.flume.*;
import org.apache.flume.conf.Configurable;
import org.apache.flume.sink.AbstractSink;

import java.io.PrintStream;
import java.util.Map;

//  Write out the data to the specified path 
public class AuthSink extends AbstractSink implements Sink, Configurable {
    

    private String path;
    private PrintStream ps;

    @Override
    public void configure(Context context) {
    
        //  Gets the specified path 
        String path = context.getString("path");
        if (path == null)
            throw new IllegalArgumentException(" Must define path attribute ！！！");
        this.path = path;
    }

    @Override
    public synchronized void start() {
    
        try {
    
            ps = new PrintStream(path + "/" + System.currentTimeMillis());
        } catch (Exception e) {
    
            e.printStackTrace();
        }
    }

    @Override
    public Status process() {
    
        //  obtain Channel
        Channel c = this.getChannel();
        //  Get transaction 
        Transaction t = c.getTransaction();
        //  Open transaction 
        t.begin();
        //  from Channel Get data in 
        Event e;
        try {
    
            while ((e = c.take()) != null) {
    
                //  obtain headers
                Map<String, String> headers = e.getHeaders();
                //  Write headers
                ps.println("headers:");
                for (Map.Entry<String, String> entry : headers.entrySet()) {
    
                    ps.println("\t" + entry.getKey() + "=" + entry.getValue());
                }
                //  obtain body
                byte[] body = e.getBody();
                //  Write body
                ps.println("body:");
                ps.println(new String(body));
            }
            //  Transaction submission 
            t.commit();
            return Status.READY;
        } catch (Exception ex) {
    
            //  Transaction rollback 
            t.rollback();
            return Status.BACKOFF;
        } finally {
    
            //  Closing transaction 
            t.close();
        }
    }

    @Override
    public synchronized void stop() {
    
        if (ps != null) {
    
            ps.close();
        }
    }

}

Package the project into cd /home/software/apache-flume-1.7.0-bin/lib Under the table of contents .
And then in cd /home/software/apache-flume-1.7.0-bin/data Create under directory conf file , The contents are as follows , Note the full path of the class and the path of the file to be written .

Selector

Processor

data Medium conf file （ Because this is played on the basis of fan out ）, Therefore, the first system node adds the following configuration information on the basis of fan out , Others are configured with fan out structure ：

Graphic description

It can also be used in sink Output to hdfs Scene , To store files by day , The configuration is as follows ：

Interceptor

Multiple interceptor chain configurations

search and Replace Interceptor configuration

Regex Filtering Interceptor configuration