Flink Kafka Sample Application (Scala)

Function Description

In a Flink application, call API of the flink-connector-kafka module to produce and consume data.

Sample Code

If you want to use FusionInsight in security mode, ensure that the kafka-clients-*.jar is obtained from the FusionInsight client directory.

Following is the main logic code of Kafka Consumer and Kafka Producer.

For the complete code, see com.huawei.bigdata.flink.examples.WriteIntoKafka and com.huawei.bigdata.flink.examples.ReadFromKafka.

//Code of producer 
object WriteIntoKafka {

     def main(args: Array[String]) {
     //Print the reference command of flink run.
       System.out.println("use command as: ")

       System.out.println("./bin/flink run --class com.huawei.bigdata.flink.examples.WriteIntoKafka" +

         " /opt/test.jar --topic topic-test -bootstrap.servers 10.91.8.218:9092")

       System.out.println
("******************************************************************************************")

       System.out.println("<topic> is the kafka topic name")

       System.out.println("<bootstrap.servers> is the ip:port list of brokers")

       System.out.println
("******************************************************************************************")
       //Build the execution environment.
       val env = StreamExecutionEnvironment.getExecutionEnvironment
       //Configure the parallelism.
       env.setParallelism(1)
       //Parse the execution parameter.
       val paraTool = ParameterTool.fromArgs(args)
       //Build the StreamGraph and wirte data generated by customized source into Kafka
       val messageStream: DataStream[String] = env.addSource(new SimpleStringGenerator)

       messageStream.addSink(new FlinkKafkaProducer(

         paraTool.get("topic"), new SimpleStringSchema, paraTool.getProperties))
       //Call execute to trigger the execution.
       env.execute

     }

   }


   //Customize source to continuously generate messages every one second.
   class SimpleStringGenerator extends SourceFunction[String] {

     var running = true

     var i = 0



     override def run(ctx: SourceContext[String]) {

       while (running) {

         ctx.collect("element-" + i)

         i += 1

         Thread.sleep(1000)

       }

     }



     override def cancel() {

       running = false

     }

   } 

//consumer code
object ReadFromKafka {

     def main(args: Array[String]) {
     //Print the reference command of flink run.
       System.out.println("use command as: ")

       System.out.println("./bin/flink run --class com.huawei.bigdata.flink.examples.ReadFromKafka" +

         " /opt/test.jar --topic topic-test -bootstrap.servers 10.91.8.218:9092")

       System.out.println
("******************************************************************************************")

       System.out.println("<topic> is the kafka topic name")

       System.out.println("<bootstrap.servers> is the ip:port list of brokers")

       System.out.println
("******************************************************************************************")


      //Build the execution environment
       val env = StreamExecutionEnvironment.getExecutionEnvironment
      //Configure the parallelism.
       env.setParallelism(1)
      //Parse the execution parameter.
       val paraTool = ParameterTool.fromArgs(args)
      //Build the StreamGraph, read data from Kafka and print the result in another row.
       val messageStream = env.addSource(new FlinkKafkaConsumer(

         paraTool.get("topic"), new SimpleStringSchema, paraTool.getProperties))

       messageStream

         .map(s => "Flink says " + s + System.getProperty("line.separator")).print()
      //Call execute to trigger the execution
       env.execute()

     }

   }