DataStream API Tutorial

Apache Flink offers a DataStream API for building robust, stateful streaming applications. It provides fine-grained control over state and time, which allows for the implementation of advanced event-driven systems. In this step-by-step guide, you’ll learn how to build a simple streaming application with PyFlink and the DataStream API.

What Will You Be Building?

In this tutorial, you will learn how to write a simple Python DataStream job. The pipeline will read data from a non-empty collection and write the results to the local file system.

Prerequisites

This walkthrough assumes that you have some familiarity with Python, but you should be able to follow along even if you come from a different programming language.

Help, I’m Stuck!

If you get stuck, check out the community support resources. In particular, Apache Flink’s user mailing list consistently ranks as one of the most active of any Apache project and a great way to get help quickly.

How To Follow Along

If you want to follow along, you will require a computer with:

  • Java 8 or 11
  • Python 3.5, 3.6 or 3.7

Using Python DataStream API requires installing PyFlink, which is available on PyPI and can be easily installed using pip.

  1. $ python -m pip install apache-flink

Once PyFlink is installed, you can move on to write a Python DataStream job.

Writing a Flink Python DataStream API Program

DataStream API applications begin by declaring an execution environment (StreamExecutionEnvironment), the context in which a streaming program is executed. This is what you will use to set the properties of your job (e.g. default parallelism, restart strategy), create your sources and finally trigger the execution of the job.

  1. env = StreamExecutionEnvironment.get_execution_environment()
  2. env.set_parallelism(1)

Once a StreamExecutionEnvironment is created, you can use it to declare your source. Sources ingest data from external systems, such as Apache Kafka, Rabbit MQ, or Apache Pulsar, into Flink Jobs.

To keep things simple, this walkthrough uses a source that is backed by a collection of elements.

  1. ds = env.from_collection(
  2.     collection=[(1, 'aaa'), (2, 'bbb')],
  3.     type_info=Types.ROW([Types.INT(), Types.STRING()]))

This creates a data stream from the given collection, with the same type as that of the elements in it (here, a ROW type with a INT field and a STRING field).

You can now perform transformations on this data stream, or just write the data to an external system using a sink. This walkthrough uses the StreamingFileSink sink connector to write the data into a file in the /tmp/output directory.

  1. ds.add_sink(StreamingFileSink
  2.     .for_row_format('/tmp/output', SimpleStringEncoder())
  3.     .build())

The last step is to execute the actual PyFlink DataStream API job. PyFlink applications are built lazily and shipped to the cluster for execution only once fully formed. To execute an application, you simply call env.execute(job_name).

  1. env.execute("tutorial_job")

The complete code so far:

  1. from pyflink.common.serialization import SimpleStringEncoder
  2. from pyflink.common.typeinfo import Types
  3. from pyflink.datastream import StreamExecutionEnvironment
  4. from pyflink.datastream.connectors import StreamingFileSink
  7. def tutorial():
  8.     env = StreamExecutionEnvironment.get_execution_environment()
  9.     env.set_parallelism(1)
  10.     ds = env.from_collection(
  11.         collection=[(1, 'aaa'), (2, 'bbb')],
  12.         type_info=Types.ROW([Types.INT(), Types.STRING()]))
  13.     ds.add_sink(StreamingFileSink
  14.                 .for_row_format('/tmp/output', SimpleStringEncoder())
  15.                 .build())
  16.     env.execute("tutorial_job")
  19. if __name__ == '__main__':
  20.     tutorial()

Executing a Flink Python DataStream API Program

Now that you defined your PyFlink program, you can run it! First, make sure that the output directory doesn’t exist:

  1. rm -rf /tmp/output

Next, you can run the example you just created on the command line:

  1. $ python datastream_tutorial.py

The command builds and runs your PyFlink program in a local mini cluster. You can alternatively submit it to a remote cluster using the instructions detailed in Job Submission Examples.

Finally, you can see the execution result on the command line:

  1. $ find /tmp/output -type f -exec cat {} \;
  2. 1,aaa
  3. 2,bbb

This walkthrough gives you the foundations to get started writing your own PyFlink DataStream API programs. To learn more about the Python DataStream API, please refer to Flink Python API Docs for more details.