flink1.17 实现 udf scalarFunctoin get_json_object 支持 非标准化json

特色

相比官方的json_value,该函数支持非标准化json,比如v是个object,但是非标准json会外套一层引号,内部有反引号.

eg: {"kkkk2": "{"kkkk1":"vvvvvvv"}" }

支持value为 100L 这种java格式的bigint. {"k":999L}

基于jsonPath 方便,可以获取多层级内部值

代码实现

import com.alibaba.fastjson2.{JSONPath, JSONReader}
import org.apache.flink.table.functions.ScalarFunction

import scala.util.Try
import org.apache.flink.configuration.{Configuration, RestOptions}
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.table.api.Expressions.row
import org.apache.flink.table.api._
import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment

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

    val conf = new Configuration
    conf.setInteger(RestOptions.PORT, 38080)
    val env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(conf)
    val tEnv = Strea