Skip to content
A collection of examples using flinks new python API
Branch: master
Clone or download
Latest commit 8dd32de Jun 8, 2018
Type Name Latest commit message Commit time
Failed to load latest commit information.
data_enrichment Remove explicit type definition Jun 19, 2016
images added images of plans and tested against newest master branch Dec 5, 2015
mandelbrot Remove explicit type definition Jun 19, 2016
mean_values tested w/ recent release, worked. Dec 29, 2017
template_example added images of plans and tested against newest master branch Dec 5, 2015
trending_hashtags tested w/ recent release, worked. Dec 29, 2017
word_count Merge pull request #3 from smferguson/fix_sysargv_lookup Jun 8, 2018
.gitignore Properly attribute Apache trademarks Jul 7, 2016 tested w/ recent release, worked. Dec 29, 2017

Python Flink™ Examples

Flink UI

A collection of examples using Apache Flink™'s new python API. To set up your local environment with the latest Flink build, see the guide:

The examples here use the v0.10.0 python API, and are meant to serve as demonstrations of simple use cases. Currently the python API supports a portion of the DataSet API, which has a similar functionality to Spark, from the user's perspective.

To run the examples, I've included a runner script at the top level with methods for each example, simply add in the path to your pyflink script and you should be good to go (as long as you have a flink cluster running locally).

The currently included examples are:


A listing of the examples and their resultant flink plans are included here.

Word Count

An extremely simple analysis program uses a source from a simple string, counts the occurrences of each word and outputs to a file on disk (using the overwrite functionality).

Word Count Plan

Trending Hashtags

A very similar example to word count, but includes a filter step to only include hashtags, and different source/sinks. The input data in this case is read off of disk, and the output is written as a csv. The file is generated dynamically at run time, so you can play with different volumes of tweets to get an idea of Flink's scalability and performance.

Trending Hashtags Plan

Data Enrichment

In this example, we have row-wise json in one file, with an attribute field that refers to a csv dimension table with colors. So we load both datasets in, convert the json data into a ordered and typed tuple, and join then two together to get a nice dataset of cars and their colors.

Data Enrichment Plan

Mean Values

Takes in a csv with two columns and finds the mean of each column, using a custom reducer function. Afterwards, it formats a string nicely with the output and dumps that onto disk.

Mean Values Plan

Mandelbrot Set

Creates a Mandelbrot set from a set of candidates. Inspired by this post

Mandelbrot Plan


A quick listing of high level features, and the examples that include them

Text data-source (read_text)

  • trending hashtags
  • data enrichment

CSV data-source (read_csv)

  • data enrichment
  • mean values
  • mandelbrot

String data-source (from_elements)

  • word count

Text output (write_text)

  • word count
  • data enrichment
  • mean values
  • mandelbrot

CSV output (write_csv)

  • trending hashtags

Log to stdout output (output)

Transformations: Map

  • word count
  • trending hashtags
  • data enrichment
  • mean values
  • mandelbrot

Transformations: FlatMap

  • word count
  • trending hashtags

Transformations: MapPartition

Transformations: Filter

  • trending hashtags
  • mandelbrot

Transformations: Reduce

  • mean values

Transformations: ReduceGroup

  • word count
  • trending hashtags

Transformations: Join

  • data enrichment

Transformations: CoGroup

Transformations: Cross

Transformations: Union

Gotchas We've Found

As we go through the process of making these examples in an extremely young library, we run across quirks, that we will mention here, and if appropriate report as bugs (we will take these down once they are fixed if they are bugs).

Using os.path to set file paths dynamically

There is a tendency to want to write code without hard-coded paths. So we may include the path to the output file in the word count example as:

import os
output_path = os.path.dirname(os.path.abspath(__file__)) + os.sep + 'output_file.txt'

But this doesn't seem to work, because some part of how pyflink is executing the python code moves it, so the abspath term evaluates to some temp directory.


Apache®, Apache Flink™, Flink™, and the Apache feather logo are trademarks of The Apache Software Foundation.

You can’t perform that action at this time.