Flink tableenv

tableEnv.kt

    private val tableEnv: StreamTableEnvironment by lazy {
        val tableEnvConfigMap = mapOf(
            "lookup.cache" to "PARTIAL",
            "lookup.partial-cache.max-rows" to "50000",
            "lokup.partial-cache.expire-after-access" to "60min"
        )
        val tableEnvSetting = EnvironmentSettings.Builder()
            .withConfiguration(Configuration.fromMap(tableEnvConfigMap))
            .build()

        val tableEnv = StreamTableEnvironment.create(env, tableEnvSetting)

        val catalog = JdbcCatalog(
            "catalogName",
            "dbname",
            config.database.mysql.username,
            config.database.mysql.password,
            config.database.mysql.url,

            )

        tableEnv.registerCatalog("catalogName", catalog)

        tableEnv.useCatalog("catalogName")

        tableEnv
    }