Configure Kafka as the message source, i.e., pulling data from Kafka into a specified sink as a consumer. The main configuration items are as follows:

Sample:

tier1.sources.source1.type = org.apache.flume.source.kafka.KafkaSource 
tier1.sources.source1.channels = channel1
tier1.sources.source1.batchSize = 5000
tier1.sources.source1.batchDurationMillis = 2000
tier1.sources.source1.kafka.bootstrap.servers = localhost:9092
tier1.sources.source1.kafka.topics = test1, test2
tier1.sources.source1.kafka.consumer.group.id = custom.g.id

For more information, please visit Apache Flume's official website.

Configure Kafka as the message receiver, i.e., pushing data into the Kafka server as a producer for subsequent operations. The main configuration items are as follows:

Sample:

a1.sinks.k1.channel = c1
a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink
a1.sinks.k1.kafka.topic = mytopic
a1.sinks.k1.kafka.bootstrap.servers = localhost:9092
a1.sinks.k1.kafka.flumeBatchSize = 20
a1.sinks.k1.kafka.producer.acks = 1

For more information, please visit Apache Flume's official website.

Step 1. Get the CKafka instance access address

1. Log in to the CKafka console.
2. Select Instance List on the left sidebar and click the ID of an instance to enter the instance basic information page.
3. On the instance basic information page, get the instance access address in the Access Mode module.
   

Step 2. Create a topic

1. On the instance basic information page, select the Topic Management tab on the top.
2. On the topic management page, click Create to create a topic named flume_test.
   

Step 3. Configure Flume

1. Download the Apache Flume toolkit and decompress it.

2. Write the configuration file flume-kafka-sink.properties. Below is a simple demo (configured in the conf folder in the extracted directory). If there is no special requirement, simply replace your own instance IP and topic in the configuration file. The source used in this example is tail -F flume-test, which represents the information newly added in the file.
   

3. Run the following command to start Flume:
   

   ./bin/flume-ng agent -n agentckafka -c conf -f conf/flume-kafka-sink.properties
   

4. Write messages to the flume-test file. At this time, the messages will be written by Flume to CKafka.
   

5. Start the CKafka client for consumption.
   

   ./kafka-console-consumer.sh --bootstrap-server xx.xx.xx.xx:xxxx --topic flume_test --from-beginning --new-consumer
   

   Note

   Enter the access address of the CKafka instance just created as bootstrap-server and the name of the topic just created as topic.

   
   You can see that the messages have been consumed.
   

Step 1. Get the CKafka instance access address

1. Log in to the CKafka console.
2. Select Instance List on the left sidebar and click the ID of an instance to enter the instance basic information page.
3. On the instance basic information page, get the instance access address in the Access Mode module.
   

Step 2. Create a topic

1. On the instance basic information page, select the Topic Management tab on the top.
2. On the topic management page, click Create to create a topic named flume_test.
   

Step 3. Configure Flume

1. Download the Apache Flume toolkit and decompress it.
2. Write the configuration file flume-kafka-source.properties. Below is a simple demo (configured in the conf folder in the extracted directory). If there is no special requirement, simply replace your own instance IP and topic in the configuration file. The sink used in this example is logger.
   
3. Run the following command to start Flume:
   

   ./bin/flume-ng agent -n agentckafka -c conf -f conf/flume-kafka-source.properties
   

4. View the logger output (the default path is logs/flume.log).