Help Center/ MapReduce Service/ Component Operation Guide (Normal)/ Using Flume/ Configuring a Non-Encrypted Flume Data Collection Task/ Using Flume Client to Collect Logs from Kafka to HDFS
Updated on 2024-10-08 GMT+08:00
View PDF
Share

Using Flume Client to Collect Logs from Kafka to HDFS

Scenario

This section describes how to use the Flume client to collect logs from the topic list (test1) of the Kafka client and save them to the /flume/test directory on HDFS.

This section applies to MRS 3.x or later.

By default, the cluster network environment is secure and the SSL authentication is not enabled during the data transmission process. For details about how to use the encryption mode, see Configuring an Encrypted Flume Data Collection Task.

Prerequisites

  • The Flume client has been installed.
  • The cluster has been installed, including the HDFS, Kafka, and Flume services.
  • You have created user flume_hdfs and authorized the HDFS directory and data to be operated during log verification.
  • The network environment of the cluster is secure.

Procedure

  1. On FusionInsight Manager, choose System > User and choose More > Download Authentication Credential to download the Kerberos certificate file of user flume_hdfs and save it to the local host.

    Figure 1 Downloading the authentication credential

  2. Configure the client parameters of the Flume role.

    1. Use the Flume configuration tool on FusionInsight Manager to configure the Flume role server parameters and generate a configuration file.
      1. Log in to FusionInsight Manager and choose Cluster > Services. On the page that is displayed, choose Flume. On the displayed page, click the Configuration Tool tab.
        Figure 2 Choosing Configuration Tool
      2. Set Agent Name to client. Select the source, channel, and sink to be used, drag them to the GUI on the right, and connect them.

        For example, use Kafka Source, File Channel, and HDFS Sink, as shown in Figure 3.

        Figure 3 Example for the Flume configuration tool
      3. Double-click the source, channel, and sink. Set corresponding configuration parameters by referring to Table 1 based on the actual environment.
        • If you want to continue using the properties.propretites file by modifying it, log in to FusionInsight Manager, choose Cluster > Name of the desired cluster > Services. On the page that is displayed, choose Flume. On the displayed page, click the Configuration Tool tab, click Import, import the file, and modify the configuration items related to non-encrypted transmission.
        • It is recommended that the numbers of Sources, Channels, and Sinks do not exceed 40 during configuration file import. Otherwise, the response time may be very long.
        Table 1 Parameters to be modified for the Flume role client

        Parameter

        Description

        Example Value

        Name

        The value must be unique and cannot be left blank.

        test

        kafka.topics

        Specifies the subscribed Kafka topic list, in which topics are separated by commas (,). This parameter cannot be left blank.

        test1

        kafka.consumer.group.id

        Specifies the data group ID obtained from Kafka. This parameter cannot be left blank.

        flume

        kafka.bootstrap.servers

        Specifies the bootstrap IP address and port list of Kafka. The default value is all Kafka lists in a Kafka cluster. If Kafka has been installed in the cluster and its configurations have been synchronized, this parameter can be left blank. This parameter is mandatory when the Flume client is used.

        192.168.101.10:21007

        batchSize

        Specifies the number of events that Flume sends in a batch (number of data pieces).

        61200

        dataDirs

        Specifies the directory for storing buffer data. The run directory is used by default. Configuring multiple directories on disks can improve transmission efficiency. Use commas (,) to separate multiple directories. If the directory is inside the cluster, the /srv/BigData/hadoop/dataX/flume/data directory can be used. dataX ranges from data1 to dataN. If the directory is outside the cluster, it needs to be independently planned.

        /srv/BigData/hadoop/data1/flume/data

        checkpointDir

        Specifies the directory for storing the checkpoint information, which is under the run directory by default. If the directory is inside the cluster, the /srv/BigData/hadoop/dataX/flume/checkpoint directory can be used. dataX ranges from data1 to dataN. If the directory is outside the cluster, it needs to be independently planned.

        /srv/BigData/hadoop/data1/flume/checkpoint

        transactionCapacity

        Specifies the transaction size, that is, the number of events in a transaction that can be processed by the current Channel. The size cannot be smaller than the batchSize of Source. Setting the same size as batchSize is recommended.

        61200

        hdfs.path

        Specifies the HDFS data write directory. This parameter cannot be left blank.

        hdfs://hacluster/flume/test

        hdfs.filePrefix

        Specifies the file name prefix after data is written to HDFS.

        TMP_

        hdfs.batchSize

        Specifies the maximum number of events that can be written to HDFS once.

        61200

        hdfs.kerberosPrincipal

        Specifies the Kerberos authentication user, which is mandatory in security versions. This configuration is required only in security clusters.

        flume_hdfs

        hdfs.kerberosKeytab

        Specifies the keytab file path for Kerberos authentication, which is mandatory in security versions. This configuration is required only in security clusters.

        /opt/test/conf/user.keytab

        NOTE:

        Obtain the user.keytab file from the Kerberos certificate file of the user flume_hdfs. In addition, ensure that the user who installs and runs the Flume client has the read and write permissions on the user.keytab file.

        hdfs.useLocalTimeStamp

        Specifies whether to use the local time. Possible values are true and false.

        true
      4. Click Export to save the properties.properties configuration file to the local.
    2. Upload the properties.properties file to flume/conf/ in the Flume client installation directory.
    3. To connect the Flume client to the HDFS, you need to add the following configuration:
      1. Download the Kerberos certificate of account flume_hdfs and obtain the krb5.conf configuration file. Upload the configuration file to the fusioninsight-flume-1.9.0/conf/ directory on the node where the client is installed.
      2. In fusioninsight-flume-1.9.0/conf/, create the jaas.conf configuration file.

        vi jaas.conf

        KafkaClient {
com.sun.security.auth.module.Krb5LoginModule required
useKeyTab=true
keyTab="/opt/test/conf/user.keytab"
principal="flume_hdfs@<System domain name>"
useTicketCache=false
storeKey=true
debug=true;
};

        Values of keyTab and principal vary depending on the actual situation.

      3. Obtain configuration files core-site.xml and hdfs-site.xml from /opt/FusionInsight_Cluster_<Cluster ID>_Flume_ClientConfig/Flume/config and upload them to fusioninsight-flume-1.9.0/conf/.
    4. Go to fusioninsight-flume-1.9.0/bin in the installation directory of the client node and run the following command to restart the Flume process:

      ./flume-manage.sh restart

  1. Verify log transmission.

    1. Log in to FusionInsight Manager as a user who has the management permission on HDFS. For details, see Accessing FusionInsight Manager (MRS 3.x or Later). Choose Cluster > Services > HDFS. On the page that is displayed, click the NameNode(Node name,Active) link next to NameNode WebUI to go to the HDFS web UI. On the displayed page, choose Utilities > Browse the file system.
    2. Check whether the data is generated in the /flume/test directory on the HDFS.
      Figure 4 Checking HDFS directories and files