apache flink - Filesystem connector produces no output & streamEnvironment.executes() throws "no operators defined"

Question

I have following code that wants to write the data generated with datagen to a file, but when I run the application, no target directory is created, and no data is written.

When I add env.execute() at the end of the code, it complains that No operators defined in streaming topology. Cannot execute.

I would ask how to make the application work, thanks.

test("insert into table") {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    val tenv = StreamTableEnvironment.create(env)
    val ddl =
      """
      create temporary table abc(
       name STRING,
       age INT
      ) with (
        'connector' = 'datagen'
      )
      """.stripMargin(' ')

    tenv.executeSql(ddl)

    val sql =
      """
        select * from abc
      """.stripMargin(' ')

    val sinkDDL =
      s"""
      create temporary table xyz(
       name STRING,
       age INT
      ) with (
       'connector' = 'filesystem',
       'path' = 'D:\${System.currentTimeMillis()}-csv' ,
       'format' = 'csv'

      )
      """.stripMargin(' ')

    tenv.executeSql(sinkDDL)

    val insertInSQL =
      """
      insert into xyz
      select name, age from abc
      """.stripMargin(' ')

    tenv.executeSql(insertInSQL)


//    env.execute()


  }

Answer 1

I think you should have UDF in table execution, see

https://ci.apache.org/projects/flink/flink-docs-stable/dev/table/functions/udfs.html#table-functions

You could see the example, write the function and insert it into your sql pipeline, this works as the "operator" in your error msg.