11.5.2、Canal

object Demo02Canal {

  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    val properties = new Properties()
    properties.setProperty("bootstrap.servers", "master:9092")
    properties.setProperty("group.id", "test1")
    //创建kafka的消费者
    val flinkKafkaCusumor = new FlinkKafkaConsumer[String]("student.order", new SimpleStringSchema(), properties)
    flinkKafkaCusumor.setStartFromEarliest()

    val canalDS: DataStream[String] = env.addSource(flinkKafkaCusumor)

    val orderDS: DataStream[(String, Double)] = canalDS.map(line => {
      //将一个json的字符串转成JSON的对象数据
      val jsonOBJ: JSONObject = JSON.parseObject(line)
      val datas: JSONArray = jsonOBJ.getJSONArray("data")
      val data = datas.getJSONObject(0)

      val id = data.getString("id")
      val order_id = data.getString("order_id")
      val amount = data.getString("amount").toDouble
      val create_time = data.getString("create_time")
      //获取类型
      val t = jsonOBJ.getString("type")
      var acc = 0.0

      t match {
        case "UPDATE" =>
          val oldJson = jsonOBJ.getJSONArray("old").getJSONObject(0)
          val oldAmount = oldJson.getString("amount").toDouble
          acc = amount - oldAmount
        case "INSERT" =>
          acc = amount
        case "DELETE" =>
          acc = -amount
      }
      (id, amount)
    })
    orderDS.keyBy(_._1)
        .sum(1)
        .print()

    env.execute()
  }

}