Using Flume from Scratch

Scenario

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

Prerequisites

• A streaming cluster with Kerberos authentication enabled has been created.
• The Flume client has been installed in a directory, for example, /opt/Flumeclient, on the node where logs are generated. For details about how to install the Flume client, see "Using Flume" > "Installing the Flume Client" in MapReduce Service Component Operation Guide. The client directory in the following operations is only an example. Change it to the actual installation directory.
• The streaming cluster can properly communicate with the node where logs are generated.

Using the Flume Client (Versions Earlier Than MRS 3.x)

You do not need to perform 2 to 6 for a normal cluster.

1. Install the client.
2. Copy the configuration file of the authentication server from the Master1 node to the Flume client installation directory/fusioninsight-flume-Flume component version number/conf directory on the node where the Flume client resides.

   The full file path is ${BIGDATA_HOME}/MRS_Current/1_X_KerberosClient/etc/kdc.conf.

   In the preceding paths, X indicates a random number. Change it based on the site requirements. The file must be saved by the user who installs the Flume client, for example, user root.

3. Check the service IP address of any node where the Flume role is deployed.
   Log in to the cluster details page, choose Name of the desired cluster > Components > Flume > Instances, and check the service IP address of any node where the Flume role is deployed.

   

   If the Components tab is unavailable, complete IAM user synchronization first. (On the Dashboard page, click Synchronize on the right side of IAM User Sync to synchronize IAM users.)

4. Copy the user authentication file from this node to the Flume client installation directory/fusioninsight-flume-Flume component version number/conf directory on the Flume client node.

   The full file path is ${BIGDATA_HOME}/MRS_XXX/install/FusionInsight-Flume-Flume component version number/flume/conf/flume.keytab.

   In the preceding paths, XXX indicates the product version number. Change it based on the site requirements. The file must be saved by the user who installs the Flume client, for example, user root.

5. Copy the jaas.conf file from this node to the conf directory on the Flume client node.

   The full file path is ${BIGDATA_HOME}/MRS_Current/1_X_Flume/etc/jaas.conf.

   In the preceding path, X indicates a random number. Change it based on the site requirements. The file must be saved by the user who installs the Flume client, for example, user root.

6. Log in to the Flume client node and go to the client installation directory. Run the following command to modify the file:

   vi conf/jaas.conf

   Change the full path of the user authentication file defined by keyTab to the Flume client installation directory/fusioninsight-flume-Flume component version number/conf saved in 4, and save the modification and exit.

7. Run the following command to modify the flume-env.sh configuration file of the Flume client:

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

   Add the following information after -XX:+UseCMSCompactAtFullCollection:

   -Djava.security.krb5.conf=Flume client installation directory/fusioninsight-flume-1.9.0/conf/kdc.conf -Djava.security.auth.login.config=Flume client installation directory/fusioninsight-flume-1.9.0/conf/jaas.conf -Dzookeeper.request.timeout=120000

   For example, "-XX:+UseCMSCompactAtFullCollection -Djava.security.krb5.conf=Flume client installation directory/fusioninsight-flume-Flume component version number/conf/kdc.conf -Djava.security.auth.login.config=Flume client installation directory/fusioninsight-flume-Flume component version number/conf/jaas.conf -Dzookeeper.request.timeout=120000"

   Change Flume client installation directory to the actual installation directory. Then save and exit.

8. Assume that the Flume client installation path is /opt/FlumeClient. Run the following command to restart the Flume client:

   cd /opt/FlumeClient/fusioninsight-flume-Flume component version number/bin

   ./flume-manage.sh restart

9. Run the following command to modify the properties.properties configuration file of the Flume client:

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

   Add the following information to the file:

   #########################################################################################
   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 = PATH
   client.sources.static_log_source.fileSuffix = .COMPLETED
   client.sources.static_log_source.ignorePattern = ^$
   client.sources.static_log_source.trackerDir = PATH
   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 = PATH
   client.channels.static_log_channel.checkpointDir = 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 = 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 = 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

   Modify the following parameters as required. Then save and exit the file.

   • spoolDir
   • trackerDir
   • dataDirs
   • checkpointDir
   • topic

     If the topic does not exist in Kafka, the topic is automatically created by default.

   • kafka.bootstrap.servers

     By default, the port for a security cluster is port 21007 and that for a normal cluster is port 9092.

   • kafka.security.protocol

     Set this parameter to SASL_PLAINTEXT for a security cluster and PLAINTEXT for a normal cluster.

   • 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.

     You can check ${BIGDATA_HOME}/MRS_Current/1_X_Broker/etc/server.properties on the node where the broker instance resides.

     In the preceding paths, X indicates a random number. Change it based on the site requirements. The file must be saved by the user who installs the Flume client, for example, user root.

10. The Flume client automatically loads the information in the properties.properties file.

    After new log files are generated in the directory specified by spoolDir, the logs will be sent to Kafka producers and can be consumed by Kafka consumers.

Using the Flume Client (MRS 3.x or Later)

You do not need to perform 2 to 6 for a normal cluster.

1. Install the client.
2. Copy the configuration file of the authentication server from the Master1 node to the Flume client installation directory/fusioninsight-flume-Flume component version number/conf directory on the node where the Flume client resides.

   The full file path is ${BIGDATA_HOME}/FusionInsight_Current/1_X_KerberosClient/etc/kdc.conf. In the preceding path, X indicates a random number. Change it based on the site requirements. The file must be saved by the user who installs the Flume client, for example, user root.

3. Check the service IP address of any node where the Flume role is deployed.

   Log in to FusionInsight Manager. For details, see Accessing FusionInsight Manager (MRS 3.x or Later). Choose Cluster > Services > Flume > Instance. Check the service IP address of any node where the Flume role is deployed.

   

   If the Components tab is unavailable, complete IAM user synchronization first. (On the Dashboard page, click Synchronize on the right side of IAM User Sync to synchronize IAM users.)

4. Copy the user authentication file from this node to the Flume client installation directory/fusioninsight-flume-Flume component version number/conf directory on the Flume client node.

   The full file path is ${BIGDATA_HOME}/FusionInsight_Porter_XXX/install/FusionInsight-Flume-Flume component version number/flume/conf/flume.keytab.

   In the preceding paths, XXX indicates the product version number. Change it based on the site requirements. The file must be saved by the user who installs the Flume client, for example, user root.

5. Copy the jaas.conf file from this node to the conf directory on the Flume client node.

   The full file path is ${BIGDATA_HOME}/FusionInsight_Current/1_X_Flume/etc/jaas.conf.

   In the preceding path, X indicates a random number. Change it based on the site requirements. The file must be saved by the user who installs the Flume client, for example, user root.

6. Log in to the Flume client node and go to the client installation directory. Run the following command to modify the file:

   vi conf/jaas.conf

   Change the full path of the user authentication file defined by keyTab to the Flume client installation directory/fusioninsight-flume-Flume component version number/conf saved in 4, and save the modification and exit.

7. Run the following command to modify the flume-env.sh configuration file of the Flume client:

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

   Add the following information after -XX:+UseCMSCompactAtFullCollection:

   -Djava.security.krb5.conf=Flume client installation directory/fusioninsight-flume-1.9.0/conf/kdc.conf -Djava.security.auth.login.config=Flume client installation directory/fusioninsight-flume-1.9.0/conf/jaas.conf -Dzookeeper.request.timeout=120000

   For example, "-XX:+UseCMSCompactAtFullCollection -Djava.security.krb5.conf=Flume client installation directory/fusioninsight-flume-Flume component version number/conf/kdc.conf -Djava.security.auth.login.config=Flume client installation directory/fusioninsight-flume-Flume component version number/conf/jaas.conf -Dzookeeper.request.timeout=120000"

   Change Flume client installation directory to the actual installation directory. Then save and exit.

8. Assume that the Flume client installation path is /opt/FlumeClient. Run the following command to restart the Flume client:

   cd /opt/FlumeClient/fusioninsight-flume-Flume component version number/bin

   ./flume-manage.sh restart

9. Run the following command to modify the properties.properties configuration file of the Flume client:

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

   Add the following information to the file:

   #########################################################################################
   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 = PATH
   client.sources.static_log_source.fileSuffix = .COMPLETED
   client.sources.static_log_source.ignorePattern = ^$
   client.sources.static_log_source.trackerDir = PATH
   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 = PATH
   client.channels.static_log_channel.checkpointDir = 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 = 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 = 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

   Modify the following parameters as required. Then save and exit the file.

   • spoolDir
   • trackerDir
   • dataDirs
   • checkpointDir
   • topic

     If the topic does not exist in Kafka, the topic is automatically created by default.

   • kafka.bootstrap.servers

     By default, the port for a security cluster is port 21007 and that for a normal cluster is port 9092.

   • kafka.security.protocol

     Set this parameter to SASL_PLAINTEXT for a security cluster and PLAINTEXT for a normal cluster.

   • 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.

     For details, check ${BIGDATA_HOME}/FusionInsight_Current/1_X_Broker/etc/server.properties on the node where the broker instance resides.

     In the preceding paths, X indicates a random number. Change it based on the site requirements. The file must be saved by the user who installs the Flume client, for example, user root.

10. The Flume client automatically loads the information in the properties.properties file.

    After new log files are generated in the directory specified by spoolDir, the logs will be sent to Kafka producers and can be consumed by Kafka consumers.