code: def word_count(input_path, output_path):
      env = StreamExecutionEnvironment.get_execution_environment()
      env.set_runtime_mode(RuntimeExecutionMode.BATCH)
      # write all the data to one file
      env.set_parallelism(1)
  
      # define the source
      if input_path is not None:
          ds = env.from_source(
              source=FileSource.for_record_stream_format(
                  StreamFormat.text_line_format(), input_path
              )
              .process_static_file_set()
              .build(),
              watermark_strategy=WatermarkStrategy.for_monotonous_timestamps(),
              source_name="file_source",
          )
      else:
          print(
              "Executing word_count example with default input data set."
          )
          print("Use --input to specify file input.")
          ds = env.from_collection(word_count_data)
  
      def split(line):
          yield from line.split()
  
      # compute word count
      ds = (
          ds.flat_map(split)
          .map(
              lambda i: (i, 1),
              output_type=Types.TUPLE([Types.STRING(), Types.INT()]),
          )
          .key_by(lambda i: i[0])
          .reduce(lambda i, j: (i[0], i[1] + j[1]))
      )
  
      # define the sink
      if output_path is not None:
          ds.sink_to(
              sink=FileSink.for_row_format(
                  base_path=output_path,
                  encoder=Encoder.simple_string_encoder(),
              )
              .with_output_file_config(
                  OutputFileConfig.builder()
                  .with_part_prefix("prefix")
                  .with_part_suffix(".ext")
                  .build()
              )
              .with_rolling_policy(
                  RollingPolicy.default_rolling_policy()
              )
              .build()
          )
      else:
          print(
              "Printing result to stdout. Use --output to specify output path."
          )
          ds.print()
  
      # submit for execution
      env.execute()
  
imports: import argparse
  import logging
  import sys
  from pyflink.common import WatermarkStrategy, Encoder, Types
  from pyflink.datastream import StreamExecutionEnvironment, RuntimeExecutionMode
  from pyflink.datastream.connectors.file_system import FileSource, StreamFormat, FileSink, OutputFileConfig, RollingPolicy
label_desc: Use image node_id 1 for calling a global function with example usage: word_count(input_path, output_path) without return types