Python REPL

Flink comes with an integrated interactive Python Shell. It can be used in a local setup as well as in a cluster setup. See the standalone resource provider page for more information about how to setup a local Flink. You can also build a local setup from source.

Note The Python Shell will run the command “python”. Please refer to the Python Table API installation guide on how to set up the Python execution environments.

To use the shell with an integrated Flink cluster, you can simply install PyFlink with PyPi and execute the shell directly:

  1. # install PyFlink
  2. $ python -m pip install apache-flink
  3. # execute the shell
  4. $ pyflink-shell.sh local

To run the shell on a cluster, please see the Setup section below.

Usage

The shell only supports Table API currently. The Table Environments are automatically prebound after startup. Use “bt_env” and “st_env” to access BatchTableEnvironment and StreamTableEnvironment respectively.

Table API

The example below is a simple program in the Python shell:

stream

  1. >>> import tempfile
  2. >>> import os
  3. >>> import shutil
  4. >>> sink_path = tempfile.gettempdir() + '/streaming.csv'
  5. >>> if os.path.exists(sink_path):
  6. ... if os.path.isfile(sink_path):
  7. ... os.remove(sink_path)
  8. ... else:
  9. ... shutil.rmtree(sink_path)
  10. >>> s_env.set_parallelism(1)
  11. >>> t = st_env.from_elements([(1, 'hi', 'hello'), (2, 'hi', 'hello')], ['a', 'b', 'c'])
  12. >>> st_env.create_temporary_table("stream_sink", TableDescriptor.for_connector("filesystem")
  13. ... .schema(Schema.new_builder()
  14. ... .column("a", DataTypes.BIGINT())
  15. ... .column("b", DataTypes.STRING())
  16. ... .column("c", DataTypes.STRING())
  17. ... .build())
  18. ... .option("path", path)
  19. ... .format(FormatDescriptor.for_format("csv")
  20. ... .option("field-delimiter", ",")
  21. ... .build())
  22. ... .build())
  23. >>> t.select("a + 1, b, c")\
  24. ... .execute_insert("stream_sink").wait()
  25. >>> # If the job runs in local mode, you can exec following code in Python shell to see the result:
  26. >>> with open(os.path.join(sink_path, os.listdir(sink_path)[0]), 'r') as f:
  27. ... print(f.read())

batch

  1. >>> import tempfile
  2. >>> import os
  3. >>> import shutil
  4. >>> sink_path = tempfile.gettempdir() + '/batch.csv'
  5. >>> if os.path.exists(sink_path):
  6. ... if os.path.isfile(sink_path):
  7. ... os.remove(sink_path)
  8. ... else:
  9. ... shutil.rmtree(sink_path)
  10. >>> b_env.set_parallelism(1)
  11. >>> t = bt_env.from_elements([(1, 'hi', 'hello'), (2, 'hi', 'hello')], ['a', 'b', 'c'])
  12. >>> st_env.create_temporary_table("batch_sink", TableDescriptor.for_connector("filesystem")
  13. ... .schema(Schema.new_builder()
  14. ... .column("a", DataTypes.BIGINT())
  15. ... .column("b", DataTypes.STRING())
  16. ... .column("c", DataTypes.STRING())
  17. ... .build())
  18. ... .option("path", path)
  19. ... .format(FormatDescriptor.for_format("csv")
  20. ... .option("field-delimiter", ",")
  21. ... .build())
  22. ... .build())
  23. >>> t.select("a + 1, b, c")\
  24. ... .execute_insert("batch_sink").wait()
  25. >>> # If the job runs in local mode, you can exec following code in Python shell to see the result:
  26. >>> with open(os.path.join(sink_path, os.listdir(sink_path)[0]), 'r') as f:
  27. ... print(f.read())

Setup

To get an overview of what options the Python Shell provides, please use

  1. pyflink-shell.sh --help

Local

To use the shell with an integrated Flink cluster just execute:

  1. pyflink-shell.sh local

Remote

To use it with a running cluster, please start the Python shell with the keyword remote and supply the host and port of the JobManager with:

  1. pyflink-shell.sh remote <hostname> <portnumber>

Yarn Python Shell cluster

The shell can deploy a Flink cluster to YARN, which is used exclusively by the shell. The shell deploys a new Flink cluster on YARN and connects the cluster. You can also specify options for YARN cluster such as memory for JobManager, name of YARN application, etc.

For example, to start a Yarn cluster for the Python Shell with two TaskManagers use the following:

  1. pyflink-shell.sh yarn -n 2

For all other options, see the full reference at the bottom.

Yarn Session

If you have previously deployed a Flink cluster using the Flink Yarn Session, the Python shell can connect with it using the following command:

  1. pyflink-shell.sh yarn

Full Reference

  1. Flink Python Shell
  2. Usage: pyflink-shell.sh [local|remote|yarn] [options] <args>...
  3. Command: local [options]
  4. Starts Flink Python shell with a local Flink cluster
  5. usage:
  6. -h,--help Show the help message with descriptions of all options.
  7. Command: remote [options] <host> <port>
  8. Starts Flink Python shell connecting to a remote cluster
  9. <host>
  10. Remote host name as string
  11. <port>
  12. Remote port as integer
  13. usage:
  14. -h,--help Show the help message with descriptions of all options.
  15. Command: yarn [options]
  16. Starts Flink Python shell connecting to a yarn cluster
  17. usage:
  18. -h,--help Show the help message with descriptions of
  19. all options.
  20. -jm,--jobManagerMemory <arg> Memory for JobManager Container with
  21. optional unit (default: MB)
  22. -nm,--name <arg> Set a custom name for the application on
  23. YARN
  24. -qu,--queue <arg> Specify YARN queue.
  25. -s,--slots <arg> Number of slots per TaskManager
  26. -tm,--taskManagerMemory <arg> Memory per TaskManager Container with
  27. optional unit (default: MB)
  28. -h | --help
  29. Prints this usage text