code: def streaming_word_count(output_path):
      t_env = TableEnvironment.create(
          EnvironmentSettings.in_streaming_mode()
      )
  
      # define the source
      # randomly select 5 words per second from a predefined list
      t_env.create_temporary_table(
          "source",
          TableDescriptor.for_connector("datagen")
          .schema(
              Schema.new_builder()
              .column("word_id", DataTypes.INT())
              .build()
          )
          .option("fields.word_id.kind", "random")
          .option("fields.word_id.min", "0")
          .option("fields.word_id.max", str(max_word_id))
          .option("rows-per-second", "5")
          .build(),
      )
      tab = t_env.from_path("source")
  
      # define the sink
      if output_path is not None:
          t_env.create_temporary_table(
              "sink",
              TableDescriptor.for_connector("filesystem")
              .schema(
                  Schema.new_builder()
                  .column("word", DataTypes.STRING())
                  .column("count", DataTypes.BIGINT())
                  .build()
              )
              .option("path", output_path)
              .format(FormatDescriptor.for_format("canal-json").build())
              .build(),
          )
      else:
          print(
              "Printing result to stdout. Use --output to specify output path."
          )
          t_env.create_temporary_table(
              "sink",
              TableDescriptor.for_connector("print")
              .schema(
                  Schema.new_builder()
                  .column("word", DataTypes.STRING())
                  .column("count", DataTypes.BIGINT())
                  .build()
              )
              .build(),
          )
  
      @udf(result_type="string")
      def id_to_word(word_id):
          return words[word_id]
  
      # compute word count
      tab.select(id_to_word(col("word_id"))).alias("word").group_by(
          col("word")
      ).select(col("word"), lit(1).count).execute_insert("sink").wait()
  
imports: import argparse
  import logging
  import sys
  from pyflink.table import TableEnvironment, EnvironmentSettings, TableDescriptor, Schema, DataTypes, FormatDescriptor
  from pyflink.table.expressions import col, lit
  from pyflink.table.udf import udf
label_desc: Use image node_id 1 for calling a global function with example usage: streaming_word_count(output_path) and returns: words