Using Flume from Scratch

Updated on 2024-11-29 GMT+08:00

View PDF

Scenario

You can use Flume to import collected log information to Kafka.

Prerequisites

A streaming cluster that contains components such as Flume and Kafka and has Kerberos authentication enabled has been created.
The streaming cluster can properly communicate with the node where logs are generated.
A human-machine user, for example, test1, has been created. The user has been added to the hadoop, yarnviewgroup, hadooppmanager, and kafkaadmin user groups as needed, and the System_administrator and default roles have been added. (If the user is created for the first time, use this username to log in to Manager and change the password.)

Using the Flume Client

Install the Flume client.

Install the Flume client in a directory, for example, /opt/Flumeclient, on the node where logs are generated by referring to Installing the Flume Client. The Flume client installation directories in the following steps are only examples. Change them to the actual installation directories.
Perform the following operations if Kerberos authentication is enabled for the cluster. Otherwise, skip these operations.
1. Download the user authentication credential.
  Log in to FusionInsight Manager and choose System > Permission > User. In the user list, locate the row containing the created user test1, click More, and select Download Authentication Credential to download the authentication credential to the local host as prompted. Decompress the package to obtain the krb5.conf and user.keytab files.
2. Copy krb5.conf and user.keytab obtained in 2.a to the Flume client installation directory/fusioninsight-flume-Flume version number/conf directory on the Flume client node.
3. Log in to the Flume client node, go to the Flume client installation directory/fusioninsight-flume-Flume version number/conf directory, and run the following command to create the jaas.conf file. Then, save the file.
  cd /opt/FlumeClient/fusioninsight-flume-Flume version number/conf/
  
  vi jaas.conf
```
KafkaClient {
com.sun.security.auth.module.Krb5LoginModule required
useKeyTab=true
keyTab="/opt/FlumeClient/fusioninsight-flume-Flume version number/conf/user.keytab"
principal="test1"
useTicketCache=false
storeKey=true
debug=true;
};
```
  NOTE:
  - keyTab: full path of the user authentication file, which is the directory for storing the user authentication file in 2.b.
  - If the user or authentication file is changed after the first authentication configuration, you need to reconfigure user authentication and restart the Flume instance.

Configure jobs based on actual service scenarios.

Some parameters can be configured on Manager.

Set the parameters in the properties.properties file. The following uses SpoolDir Source+File Channel+Kafka Sink as an example.

Run the following command on the node where the Flume client is installed. Configure and save jobs in the Flume client configuration file properties.properties based on actual service requirements.

vi Flume client installation directory/fusioninsight-flume-Flume component version number/conf/properties.properties

#########################################################################################
client.sources = static_log_source  
client.channels = static_log_channel 
client.sinks = kafka_sink
#########################################################################################
#LOG_TO_HDFS_ONLINE_1

client.sources.static_log_source.type = spooldir
client.sources.static_log_source.spoolDir = Monitoring directory
client.sources.static_log_source.fileSuffix = .COMPLETED
client.sources.static_log_source.ignorePattern = ^$
client.sources.static_log_source.trackerDir = Metadata storage path during transmission
client.sources.static_log_source.maxBlobLength = 16384
client.sources.static_log_source.batchSize = 51200
client.sources.static_log_source.inputCharset = UTF-8
client.sources.static_log_source.deserializer = LINE
client.sources.static_log_source.selector.type = replicating
client.sources.static_log_source.fileHeaderKey = file
client.sources.static_log_source.fileHeader = false
client.sources.static_log_source.basenameHeader = true
client.sources.static_log_source.basenameHeaderKey = basename
client.sources.static_log_source.deletePolicy = never

client.channels.static_log_channel.type = file
client.channels.static_log_channel.dataDirs = Data cache path. Multiple paths, separated by commas (,), can be configured to improve performance.
client.channels.static_log_channel.checkpointDir = Checkpoint storage path
client.channels.static_log_channel.maxFileSize = 2146435071
client.channels.static_log_channel.capacity = 1000000
client.channels.static_log_channel.transactionCapacity = 612000
client.channels.static_log_channel.minimumRequiredSpace = 524288000

client.sinks.kafka_sink.type = org.apache.flume.sink.kafka.KafkaSink
client.sinks.kafka_sink.kafka.topic = Topic to which data is written, for example, flume_test
client.sinks.kafka_sink.kafka.bootstrap.servers = XXX.XXX.XXX.XXX:Kafka port number,XXX.XXX.XXX.XXX:Kafka port number,XXX.XXX.XXX.XXX:Kafka port number
client.sinks.kafka_sink.flumeBatchSize = 1000
client.sinks.kafka_sink.kafka.producer.type = sync
client.sinks.kafka_sink.kafka.security.protocol = SASL_PLAINTEXT
client.sinks.kafka_sink.kafka.kerberos.domain.name = Kafka domain name. This parameter is mandatory for a security cluster, for example, hadoop.xxx.com.
client.sinks.kafka_sink.requiredAcks = 0

client.sources.static_log_source.channels = static_log_channel
client.sinks.kafka_sink.channel = static_log_channel

NOTE:

client.sinks.kafka_sink.kafka.topic: Topic to which data is written. If the topic does not exist in Kafka, it is automatically created by default.
client.sinks.kafka_sink.kafka.bootstrap.servers: List of Kafka Brokers, which are separated by commas (,). By default, the port is 21007 for a security cluster and 9092 for a normal cluster.
client.sinks.kafka_sink.kafka.security.protocol: The value is SASL_PLAINTEXT for a security cluster and PLAINTEXT for a normal cluster.
client.sinks.kafka_sink.kafka.kerberos.domain.name:
You do not need to set this parameter for a normal cluster. For a security cluster, the value of this parameter is the value of kerberos.domain.name in the Kafka cluster.

Obtain the value by checking ${BIGDATA_HOME}/FusionInsight_Current/1_X_Broker/etc/server.properties on the node where the broker instance is located.

After the parameters are configured and saved, the Flume client automatically loads the content configured in properties.properties. When new log files are generated by spoolDir, the files are sent to Kafka producers and can be consumed by Kafka consumers.