code: def word_count(input_path, output_path):
      t_env = TableEnvironment.create(
          EnvironmentSettings.in_streaming_mode()
      )
      # write all the data to one file
      t_env.get_config().set("parallelism.default", "1")
  
      # define the source
      if input_path is not None:
          t_env.create_temporary_table(
              "source",
              TableDescriptor.for_connector("filesystem")
              .schema(
                  Schema.new_builder()
                  .column("word", DataTypes.STRING())
                  .build()
              )
              .option("path", input_path)
              .format("csv")
              .build(),
          )
          tab = t_env.from_path("source")
      else:
          print(
              "Executing word_count example with default input data set."
          )
          print("Use --input to specify file input.")
          tab = t_env.from_elements(
              map(lambda i: (i,), word_count_data),
              DataTypes.ROW(
                  [DataTypes.FIELD("line", DataTypes.STRING())]
              ),
          )
  
      # 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(),
          )
  
      @udtf(result_types=[DataTypes.STRING()])
      def split(line: Row):
          for s in line[0].split():
              yield Row(s)
  
      # compute word count
      tab.flat_map(split).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.common import Row
  from pyflink.table import EnvironmentSettings, TableEnvironment, TableDescriptor, Schema, DataTypes, FormatDescriptor
  from pyflink.table.expressions import lit, col
  from pyflink.table.udf import udtf
label_desc: Use image node_id 1 for calling a global function with example usage: word_count(input_path, output_path) without return types