批处理的例子

下面的示例程序展示了Flink的不同应用程序，从简单的单词计数到图形算法。代码示例演示了Flink’s DataSet API的使用。

以下示例的完整源代码和更多示例可以在Flink源代码存储库的flink-examples-batch中找到。

运行一个例子

为了运行Flink示例，我们假设您有一个正在运行的Flink实例可用。导航中的“快速启动”和“设置”选项卡描述了启动Flink的各种方法。

最简单的方法是运行./bin/start-cluster.sh。默认情况下，它使用一个JobManager和一个TaskManager启动本地集群。

Flink的每个二进制版本都包含一个examples目录，其中包含用于该页上每个示例的jar文件。

要运行WordCount示例，发出以下命令:

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

其他示例也可以以类似的方式开始。

注意，通过使用内置数据，许多示例在运行时不传递任何参数。要使用真实数据运行WordCount，必须将路径传递给数据:

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

注意，非本地文件系统需要一个模式前缀，如hdfs://。

Word Count

WordCount是大数据处理系统的“Hello World”。它计算文本集合中单词的频率。该算法分为两个步骤:首先，将文本拆分为单个单词。其次，对单词进行分组和计数。

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

WordCount example 使用输入参数实现了上述算法:--input &lt;path&gt; --output &lt;path&gt;。作为测试数据，任何文本文件都可以。

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

WordCount example 使用输入参数实现了上述算法:--input &lt;path&gt; --output &lt;path&gt;。作为测试数据，任何文本文件都可以。

网页排名

PageRank算法计算链接定义的图中页面的“重要性”，链接从一个页面指向另一个页面。它是一种迭代图算法，即重复应用相同的计算。在每次迭代中，每个页面都将其当前的秩分布到所有相邻的页面上，并将其新秩计算为从相邻页面获得的秩的累加和。PageRank算法是由谷歌搜索引擎推广的，它利用网页的重要性对搜索查询结果进行排序。

在这个简单的例子中，PageRank是通过bulk iteration和固定数量的迭代来实现的。

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

PageRank program实现了上述示例。它需要以下参数来运行:--pages &lt;path&gt; --links &lt;path&gt; --output &lt;path&gt; --numPages &lt;n&gt; --iterations &lt;n&gt;。

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

PageRank程序实现了上述示例。它需要以下参数来运行:--pages &lt;path&gt; --links &lt;path&gt; --output &lt;path&gt; --numPages &lt;n&gt; --iterations &lt;n&gt;。

输入文件是纯文本文件，必须格式化如下:

  • 页面表示为由新行字符分隔的(长)ID。
    • 例如"1\n2\n12\n42\n63\n" 给出了5页的IDs 1、2、12、42和63。
  • 链接表示为由空格字符分隔的页面id对。链接用换行符分隔:
    • 例如"1 2\n2 12\n1 12\n42 63\n" 给四(导演)链接(1)→(2),(2)→(12),(1)→(12)和(42)→(63)。

对于这个简单的实现，要求每个页面至少有一个传入和一个传出链接(页面可以指向自己)。

连接组件

连通分量算法通过将同一连通部分中的所有顶点分配给相同的分量ID来识别一个较大图中的连通部分。与PageRank类似，连通分量是一种迭代算法。在每一步中，每个顶点将其当前的组件ID传播到它的所有邻居。如果一个顶点的组件ID小于它自己的组件ID，那么它接受来自邻居的组件ID。

此实现使用delta iteration:没有更改其组件ID的顶点不参与下一步。这将产生更好的性能，因为后面的迭代通常只处理少数离群点。

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

ConnectedComponents program 实现了上述示例。它需要以下参数来运行:--vertices &lt;path&gt; --edges &lt;path&gt; --output &lt;path&gt; --iterations &lt;n&gt;。

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

ConnectedComponents程序实现了上述示例。它需要以下参数来运行:--vertices &lt;path&gt; --edges &lt;path&gt; --output &lt;path&gt; --iterations &lt;n&gt;。

输入文件是纯文本文件，必须格式化如下:

  • 顶点表示为id，用换行符分隔。
    • 例如"1\n2\n12\n42\n63\n" 给了5个顶点(1)，(2)，(12)，(42)和(63)。
  • 边缘表示为顶点id的对，顶点id由空间字符分隔。边缘用换行符分隔:
    • 例如"1 2\n2 12\n1 12\n42 63\n"给出4个(无向)链路(1)-(2)，(2)-(12)，(1)-(12)，和(42)-(63)。