Batch Examples

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

    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 , 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:

    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://.

    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.

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

    Scala

    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.

    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 and a fixed number of iterations.

    Java

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

    Scala

    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)))
    23.   (l1, l2) => AdjacencyList(l1.sourceId, l1.targetIds ++ l2.targetIds)
    24.   }
    26. // start iteration
    27. val finalRanks = pagesWithRanks.iterateWithTermination(maxIterations) {
    28.   currentRanks =>
    29.     val newRanks = currentRanks
    30.       // distribute ranks to target pages
    31.       .join(adjacencyLists).where("pageId").equalTo("sourceId") {
    32.         (page, adjacent, out: Collector[Page]) =>
    33.         for (targetId <- adjacent.targetIds) {
    34.           out.collect(Page(targetId, page.rank / adjacent.targetIds.length))
    35.         }
    36.       }
    37.       // collect ranks and sum them up
    38.       .groupBy("pageId").aggregate(SUM, "rank")
    39.       // apply dampening factor
    40.       .map { p =>
    41.         Page(p.pageId, (p.rank * DAMPENING_FACTOR) + ((1 - DAMPENING_FACTOR) / numPages))
    42.       }
    44.     // terminate if no rank update was significant
    45.     val termination = currentRanks.join(newRanks).where("pageId").equalTo("pageId") {
    46.       (current, next, out: Collector[Int]) =>
    47.         // check for significant update
    48.         if (math.abs(current.rank - next.rank) > EPSILON) out.collect(1)
    49.     }
    51.     (newRanks, termination)
    52. }
    54. val result = finalRanks
    56. // emit result
    57. result.writeAsCsv(outputPath, "\n", " ")

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

    • 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 this simple implementation it is required that each page has at least one incoming and one outgoing link (a page can point to itself).

    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.

    Java

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

    Scala

    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).