Batch Examples

The following example programs showcase different applications of Flinkfrom simple word counting to graph algorithms. The code samples illustrate theuse of Flink’s DataSet API.

The full source code of the following and more examples can be found in the flink-examples-batch module of the Flink source repository.

Running an example

In order to run a Flink example, we assume you have a running Flink instance available. The “Quickstart” and “Setup” tabs in the navigation describe various ways of starting Flink.

The easiest way is running the ./bin/start-cluster.sh, which by default starts a local cluster with one JobManager and one TaskManager.

Each binary release of Flink contains an examples directory with jar files for each of the examples on this page.

To run the WordCount example, issue the following command:

  1. ./bin/flink run ./examples/batch/WordCount.jar

The other examples can be started in a similar way.

Note that many examples run without passing any arguments for them, by using build-in data. To run WordCount with real data, you have to pass the path to the data:

  1. ./bin/flink run ./examples/batch/WordCount.jar --input /path/to/some/text/data --output /path/to/result

Note that non-local file systems require a schema prefix, such as hdfs://.

Word Count

WordCount is the “Hello World” of Big Data processing systems. It computes the frequency of words in a text collection. The algorithm works in two steps: First, the texts are splits the text to individual words. Second, the words are grouped and counted.

  1. ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
  3. DataSet<String> text = env.readTextFile("/path/to/file");
  5. DataSet<Tuple2<String, Integer>> counts =
  6.         // split up the lines in pairs (2-tuples) containing: (word,1)
  7.         text.flatMap(new Tokenizer())
  8.         // group by the tuple field "0" and sum up tuple field "1"
  9.         .groupBy(0)
  10.         .sum(1);
  12. counts.writeAsCsv(outputPath, "\n", " ");
  14. // User-defined functions
  15. public static class Tokenizer implements FlatMapFunction<String, Tuple2<String, Integer>> {
  17.     @Override
  18.     public void flatMap(String value, Collector<Tuple2<String, Integer>> out) {
  19.         // normalize and split the line
  20.         String[] tokens = value.toLowerCase().split("\\W+");
  22.         // emit the pairs
  23.         for (String token : tokens) {
  24.             if (token.length() > 0) {
  25.                 out.collect(new Tuple2<String, Integer>(token, 1));
  26.             }   
  27.         }
  28.     }
  29. }

The WordCount example implements the above described algorithm with input parameters: —input <path> —output <path>. As test data, any text file will do.

  1. val env = ExecutionEnvironment.getExecutionEnvironment
  3. // get input data
  4. val text = env.readTextFile("/path/to/file")
  6. val counts = text.flatMap { _.toLowerCase.split("\\W+") filter { _.nonEmpty } }
  7.   .map { (_, 1) }
  8.   .groupBy(0)
  9.   .sum(1)
  11. counts.writeAsCsv(outputPath, "\n", " ")

The WordCount example implements the above described algorithm with input parameters: —input <path> —output <path>. As test data, any text file will do.

Page Rank

The PageRank algorithm computes the “importance” of pages in a graph defined by links, which point from one pages to another page. It is an iterative graph algorithm, which means that it repeatedly applies the same computation. In each iteration, each page distributes its current rank over all its neighbors, and compute its new rank as a taxed sum of the ranks it received from its neighbors. The PageRank algorithm was popularized by the Google search engine which uses the importance of webpages to rank the results of search queries.

In this simple example, PageRank is implemented with a bulk iteration and a fixed number of iterations.

  1. ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
  3. // read the pages and initial ranks by parsing a CSV file
  4. DataSet<Tuple2<Long, Double>> pagesWithRanks = env.readCsvFile(pagesInputPath)
  5.                            .types(Long.class, Double.class)
  7. // the links are encoded as an adjacency list: (page-id, Array(neighbor-ids))
  8. DataSet<Tuple2<Long, Long[]>> pageLinkLists = getLinksDataSet(env);
  10. // set iterative data set
  11. IterativeDataSet<Tuple2<Long, Double>> iteration = pagesWithRanks.iterate(maxIterations);
  13. DataSet<Tuple2<Long, Double>> newRanks = iteration
  14.         // join pages with outgoing edges and distribute rank
  15.         .join(pageLinkLists).where(0).equalTo(0).flatMap(new JoinVertexWithEdgesMatch())
  16.         // collect and sum ranks
  17.         .groupBy(0).sum(1)
  18.         // apply dampening factor
  19.         .map(new Dampener(DAMPENING_FACTOR, numPages));
  21. DataSet<Tuple2<Long, Double>> finalPageRanks = iteration.closeWith(
  22.         newRanks,
  23.         newRanks.join(iteration).where(0).equalTo(0)
  24.         // termination condition
  25.         .filter(new EpsilonFilter()));
  27. finalPageRanks.writeAsCsv(outputPath, "\n", " ");
  29. // User-defined functions
  31. public static final class JoinVertexWithEdgesMatch
  32.                     implements FlatJoinFunction<Tuple2<Long, Double>, Tuple2<Long, Long[]>,
  33.                                             Tuple2<Long, Double>> {
  35.     @Override
  36.     public void join(<Tuple2<Long, Double> page, Tuple2<Long, Long[]> adj,
  37.                         Collector<Tuple2<Long, Double>> out) {
  38.         Long[] neighbors = adj.f1;
  39.         double rank = page.f1;
  40.         double rankToDistribute = rank / ((double) neigbors.length);
  42.         for (int i = 0; i < neighbors.length; i++) {
  43.             out.collect(new Tuple2<Long, Double>(neighbors[i], rankToDistribute));
  44.         }
  45.     }
  46. }
  48. public static final class Dampener implements MapFunction<Tuple2<Long,Double>, Tuple2<Long,Double>> {
  49.     private final double dampening, randomJump;
  51.     public Dampener(double dampening, double numVertices) {
  52.         this.dampening = dampening;
  53.         this.randomJump = (1 - dampening) / numVertices;
  54.     }
  56.     @Override
  57.     public Tuple2<Long, Double> map(Tuple2<Long, Double> value) {
  58.         value.f1 = (value.f1 * dampening) + randomJump;
  59.         return value;
  60.     }
  61. }
  63. public static final class EpsilonFilter
  64.                 implements FilterFunction<Tuple2<Tuple2<Long, Double>, Tuple2<Long, Double>>> {
  66.     @Override
  67.     public boolean filter(Tuple2<Tuple2<Long, Double>, Tuple2<Long, Double>> value) {
  68.         return Math.abs(value.f0.f1 - value.f1.f1) > EPSILON;
  69.     }
  70. }

The PageRank program implements the above example.It requires the following parameters to run: —pages <path> —links <path> —output <path> —numPages <n> —iterations <n>.

  1. // User-defined types
  2. case class Link(sourceId: Long, targetId: Long)
  3. case class Page(pageId: Long, rank: Double)
  4. case class AdjacencyList(sourceId: Long, targetIds: Array[Long])
  6. // set up execution environment
  7. val env = ExecutionEnvironment.getExecutionEnvironment
  9. // read the pages and initial ranks by parsing a CSV file
  10. val pages = env.readCsvFile[Page](pagesInputPath)
  12. // the links are encoded as an adjacency list: (page-id, Array(neighbor-ids))
  13. val links = env.readCsvFile[Link](linksInputPath)
  15. // assign initial ranks to pages
  16. val pagesWithRanks = pages.map(p => Page(p, 1.0 / numPages))
  18. // build adjacency list from link input
  19. val adjacencyLists = links
  20.   // initialize lists
  21.   .map(e => AdjacencyList(e.sourceId, Array(e.targetId)))
  22.   // concatenate lists
  23.   .groupBy("sourceId").reduce {
  24.   (l1, l2) => AdjacencyList(l1.sourceId, l1.targetIds ++ l2.targetIds)
  25.   }
  27. // start iteration
  28. val finalRanks = pagesWithRanks.iterateWithTermination(maxIterations) {
  29.   currentRanks =>
  30.     val newRanks = currentRanks
  31.       // distribute ranks to target pages
  32.       .join(adjacencyLists).where("pageId").equalTo("sourceId") {
  33.         (page, adjacent, out: Collector[Page]) =>
  34.         for (targetId <- adjacent.targetIds) {
  35.           out.collect(Page(targetId, page.rank / adjacent.targetIds.length))
  36.         }
  37.       }
  38.       // collect ranks and sum them up
  39.       .groupBy("pageId").aggregate(SUM, "rank")
  40.       // apply dampening factor
  41.       .map { p =>
  42.         Page(p.pageId, (p.rank * DAMPENING_FACTOR) + ((1 - DAMPENING_FACTOR) / numPages))
  43.       }
  45.     // terminate if no rank update was significant
  46.     val termination = currentRanks.join(newRanks).where("pageId").equalTo("pageId") {
  47.       (current, next, out: Collector[Int]) =>
  48.         // check for significant update
  49.         if (math.abs(current.rank - next.rank) > EPSILON) out.collect(1)
  50.     }
  52.     (newRanks, termination)
  53. }
  55. val result = finalRanks
  57. // emit result
  58. result.writeAsCsv(outputPath, "\n", " ")

The PageRank program implements the above example.It requires the following parameters to run: —pages <path> —links <path> —output <path> —numPages <n> —iterations <n>.

Input files are plain text files and must be formatted as follows:

  • Pages represented as an (long) ID separated by new-line characters.
    • For example "1\n2\n12\n42\n63\n" gives five pages with IDs 1, 2, 12, 42, and 63.
  • Links are represented as pairs of page IDs which are separated by space characters. Links are separated by new-line characters:
    • For example "1 2\n2 12\n1 12\n42 63\n" gives four (directed) links (1)->(2), (2)->(12), (1)->(12), and (42)->(63).For this simple implementation it is required that each page has at least one incoming and one outgoing link (a page can point to itself).

Connected Components

The Connected Components algorithm identifies parts of a larger graph which are connected by assigning all vertices in the same connected part the same component ID. Similar to PageRank, Connected Components is an iterative algorithm. In each step, each vertex propagates its current component ID to all its neighbors. A vertex accepts the component ID from a neighbor, if it is smaller than its own component ID.

This implementation uses a delta iteration: Vertices that have not changed their component ID do not participate in the next step. This yields much better performance, because the later iterations typically deal only with a few outlier vertices.

  1. // read vertex and edge data
  2. DataSet<Long> vertices = getVertexDataSet(env);
  3. DataSet<Tuple2<Long, Long>> edges = getEdgeDataSet(env).flatMap(new UndirectEdge());
  5. // assign the initial component IDs (equal to the vertex ID)
  6. DataSet<Tuple2<Long, Long>> verticesWithInitialId = vertices.map(new DuplicateValue<Long>());
  8. // open a delta iteration
  9. DeltaIteration<Tuple2<Long, Long>, Tuple2<Long, Long>> iteration =
  10.         verticesWithInitialId.iterateDelta(verticesWithInitialId, maxIterations, 0);
  12. // apply the step logic:
  13. DataSet<Tuple2<Long, Long>> changes = iteration.getWorkset()
  14.         // join with the edges
  15.         .join(edges).where(0).equalTo(0).with(new NeighborWithComponentIDJoin())
  16.         // select the minimum neighbor component ID
  17.         .groupBy(0).aggregate(Aggregations.MIN, 1)
  18.         // update if the component ID of the candidate is smaller
  19.         .join(iteration.getSolutionSet()).where(0).equalTo(0)
  20.         .flatMap(new ComponentIdFilter());
  22. // close the delta iteration (delta and new workset are identical)
  23. DataSet<Tuple2<Long, Long>> result = iteration.closeWith(changes, changes);
  25. // emit result
  26. result.writeAsCsv(outputPath, "\n", " ");
  28. // User-defined functions
  30. public static final class DuplicateValue<T> implements MapFunction<T, Tuple2<T, T>> {
  32.     @Override
  33.     public Tuple2<T, T> map(T vertex) {
  34.         return new Tuple2<T, T>(vertex, vertex);
  35.     }
  36. }
  38. public static final class UndirectEdge
  39.                     implements FlatMapFunction<Tuple2<Long, Long>, Tuple2<Long, Long>> {
  40.     Tuple2<Long, Long> invertedEdge = new Tuple2<Long, Long>();
  42.     @Override
  43.     public void flatMap(Tuple2<Long, Long> edge, Collector<Tuple2<Long, Long>> out) {
  44.         invertedEdge.f0 = edge.f1;
  45.         invertedEdge.f1 = edge.f0;
  46.         out.collect(edge);
  47.         out.collect(invertedEdge);
  48.     }
  49. }
  51. public static final class NeighborWithComponentIDJoin
  52.                 implements JoinFunction<Tuple2<Long, Long>, Tuple2<Long, Long>, Tuple2<Long, Long>> {
  54.     @Override
  55.     public Tuple2<Long, Long> join(Tuple2<Long, Long> vertexWithComponent, Tuple2<Long, Long> edge) {
  56.         return new Tuple2<Long, Long>(edge.f1, vertexWithComponent.f1);
  57.     }
  58. }
  60. public static final class ComponentIdFilter
  61.                     implements FlatMapFunction<Tuple2<Tuple2<Long, Long>, Tuple2<Long, Long>>,
  62.                                             Tuple2<Long, Long>> {
  64.     @Override
  65.     public void flatMap(Tuple2<Tuple2<Long, Long>, Tuple2<Long, Long>> value,
  66.                         Collector<Tuple2<Long, Long>> out) {
  67.         if (value.f0.f1 < value.f1.f1) {
  68.             out.collect(value.f0);
  69.         }
  70.     }
  71. }

The ConnectedComponents program implements the above example. It requires the following parameters to run: —vertices <path> —edges <path> —output <path> —iterations <n>.

  1. // set up execution environment
  2. val env = ExecutionEnvironment.getExecutionEnvironment
  4. // read vertex and edge data
  5. // assign the initial components (equal to the vertex id)
  6. val vertices = getVerticesDataSet(env).map { id => (id, id) }
  8. // undirected edges by emitting for each input edge the input edges itself and an inverted
  9. // version
  10. val edges = getEdgesDataSet(env).flatMap { edge => Seq(edge, (edge._2, edge._1)) }
  12. // open a delta iteration
  13. val verticesWithComponents = vertices.iterateDelta(vertices, maxIterations, Array(0)) {
  14.   (s, ws) =>
  16.     // apply the step logic: join with the edges
  17.     val allNeighbors = ws.join(edges).where(0).equalTo(0) { (vertex, edge) =>
  18.       (edge._2, vertex._2)
  19.     }
  21.     // select the minimum neighbor
  22.     val minNeighbors = allNeighbors.groupBy(0).min(1)
  24.     // update if the component of the candidate is smaller
  25.     val updatedComponents = minNeighbors.join(s).where(0).equalTo(0) {
  26.       (newVertex, oldVertex, out: Collector[(Long, Long)]) =>
  27.         if (newVertex._2 < oldVertex._2) out.collect(newVertex)
  28.     }
  30.     // delta and new workset are identical
  31.     (updatedComponents, updatedComponents)
  32. }
  34. verticesWithComponents.writeAsCsv(outputPath, "\n", " ")

The ConnectedComponents program implements the above example. It requires the following parameters to run: —vertices <path> —edges <path> —output <path> —iterations <n>.

Input files are plain text files and must be formatted as follows:

  • Vertices represented as IDs and separated by new-line characters.
    • For example "1\n2\n12\n42\n63\n" gives five vertices with (1), (2), (12), (42), and (63).
  • Edges are represented as pairs for vertex IDs which are separated by space characters. Edges are separated by new-line characters:
    • For example "1 2\n2 12\n1 12\n42 63\n" gives four (undirected) links (1)-(2), (2)-(12), (1)-(12), and (42)-(63).