override def handle()

in quickstart-flink/quickstart-flinksql/flinksql_1.12/src/main/scala/org/apache/streampark/quickstart/flink/flinksql/MixinTableDataStreamJob.scala [25:46]

• 17 lines of code
• 1 McCabe index (conditional complexity)

  override def handle(): Unit = {

    val mySource = context.$fromCollection(
      List(
        "kafka,apapche kafka",
        "spark,spark",
        "flink,apapche flink",
        "zookeeper,apapche zookeeper",
        "hadoop,apapche hadoop"
      )
    ).map(x => {
      val array = x.split(",")
      Entity(array.head, array.last)
    })

    context.createTemporaryView("mySource", mySource)

    //kafka to table
    val table1: Table = context.from("mySource")
    table1.>>[Entity].print("stream print==>")

  }