flink实例开发-batch批处理实例

缺乏、安全感 2022-05-30 10:43 363阅读 0赞

batch批处理实例

下面的示例程序将展示flink的不同应用程序从简单的单词计数到图计算。示例代码演示使用Flink的DataSet API。

以下的全部源代码和更多的例子可以在flink源码仓库的flink-examples-batch或者flink-examples-streaming模块中看到。

  • 运行一个示例
  • Word Count 单词计数
  • Page Rank 网页排名
  • Connected Componexts 连通分支
  • Relational Query 关系查询

运行一个示例

为了运行flink程序，我们假设你已经有一个可以运行的flink集群。在快速开始和实例开发模块中详细描述了如何启动flink。

最简单的方法是运行 ./bin/start-local.sh 脚本。这将启动一个本地的JobManager。

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”入门程序。它计算单词在文本集合中的频率。

该算法有两步骤：首先，该文本被切割成单词，然后，这些单词被分组和求和。

java代码：

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

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", " ")

此链接中的wordCount example代码实现了输入参数：—input —output ， 作为测试数据，任何文件文件都可以。

Page Rank 网页排名

PageRank算法通过图中定义的链接计算网页重要性，从一个页面到另一个页面。这是一个迭代的图算法，这意味着它需要执行相同的计算。在每个迭代中，每一页分配当前排名给它所有的邻居，并从邻居节点计算其新等级。PageRank算法是由google进行普及的，主要用在计算网页的重要性排名，应用在搜索结果里面。

在这个简单示例中，PageRank实现批量得带和固定数量的迭代。

java代码：

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

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)))
  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", " ")

这个PageRank程序实现了上面的例子。它需要以下参数运行：

  1. --pages <path> --links <path> --output <path> --numPages <n> --iterations <n>

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

  • pages 使用一个long型的id进行表示，通过换行符进行分割。

  1. 例如： “1\n2\n12\n42\n63\n” 表示给了5个pages 1, 2, 12, 42, 和63.

    • links 使用一对page id进行表示，这一对page id通过空格进行分割。Links通过换行符进行分割

  2. 例如：”1 2\n2 12\n1 12\n42 63\n” 表示给了4对 links (1)->(2), (2)->(12), (1)->(12), 和(42)->(63)

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

Connected Componexts 连通分支

算法识别部分的连接组件的一个更大的图连接通过分配所有顶点在相同的连接部分组件ID。类似于PageRank,连接组件是一个迭代算法。在每一步,每个顶点传播当前所有相邻组件ID。一个顶点接受邻居的组件ID,如果它小于自己的组件ID

这个实现使用一个增量迭代:顶点并没有改变他们的组件ID不参与下一步。这收益更好的性能,因为后来的迭代通常只处理一些异常值顶点

java代码

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

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", " ")

ConnectedComponents程序实现了上面的例子。它需要以下参数运行：

  1. --vertices <path> --edges <path> --output <path> --iterations <n>

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

  • Vertices(顶点)使用id表示，使用换行符隔开
  1. 例如 “1\n2\n12\n42\n63\n” 表示指定5个顶点 (1), (2), (12), (42), and (63).
  • 顶点的边缘表示为两个id，使用空格隔开，边缘由换行符隔开
  1. 例如： “1 2\n2 12\n1 12\n42 63\n” 表示4个无向链接 (1)-(2), (2)-(12), (1)-(12), and (42)-(63).

Relational Query 关系查询

关系查询的例子假定两个表,一个订单和指定的其他与lineitem tpc - h基准决策支持。tpc - h是一个标准的数据库行业的基准。见下文说明如何生成输入数据

这个例子实现了下面的SQL查询

  1. SELECT l_orderkey, o_shippriority, sum(l_extendedprice) as revenue
  2.     FROM orders, lineitem
  3. WHERE l_orderkey = o_orderkey
  4.     AND o_orderstatus = "F"
  5.     AND YEAR(o_orderdate) > 1993
  6.     AND o_orderpriority LIKE "5%"
  7. GROUP BY l_orderkey, o_shippriority;

Flink程序，实现了上面的查询，看起来如下：

java代码：

  1. // get orders data set: (orderkey, orderstatus, orderdate, orderpriority, shippriority)
  2. DataSet<Tuple5<Integer, String, String, String, Integer>> orders = getOrdersDataSet(env);
  3. // get lineitem data set: (orderkey, extendedprice)
  4. DataSet<Tuple2<Integer, Double>> lineitems = getLineitemDataSet(env);
  6. // orders filtered by year: (orderkey, custkey)
  7. DataSet<Tuple2<Integer, Integer>> ordersFilteredByYear =
  8.         // filter orders
  9.         orders.filter(
  10.             new FilterFunction<Tuple5<Integer, String, String, String, Integer>>() {
  11.                 @Override
  12.                 public boolean filter(Tuple5<Integer, String, String, String, Integer> t) {
  13.                     // status filter
  14.                     if(!t.f1.equals(STATUS_FILTER)) {
  15.                         return false;
  16.                     // year filter
  17.                     } else if(Integer.parseInt(t.f2.substring(0, 4)) <= YEAR_FILTER) {
  18.                         return false;
  19.                     // order priority filter
  20.                     } else if(!t.f3.startsWith(OPRIO_FILTER)) {
  21.                         return false;
  22.                     }
  23.                     return true;
  24.                 }
  25.             })
  26.         // project fields out that are no longer required
  27.         .project(0,4).types(Integer.class, Integer.class);
  29. // join orders with lineitems: (orderkey, shippriority, extendedprice)
  30. DataSet<Tuple3<Integer, Integer, Double>> lineitemsOfOrders =
  31.         ordersFilteredByYear.joinWithHuge(lineitems)
  32.                             .where(0).equalTo(0)
  33.                             .projectFirst(0,1).projectSecond(1)
  34.                             .types(Integer.class, Integer.class, Double.class);
  36. // extendedprice sums: (orderkey, shippriority, sum(extendedprice))
  37. DataSet<Tuple3<Integer, Integer, Double>> priceSums =
  38.         // group by order and sum extendedprice
  39.         lineitemsOfOrders.groupBy(0,1).aggregate(Aggregations.SUM, 2);
  41. // emit result
  42. priceSums.writeAsCsv(outputPath);

这个Relational Query实现了上面的查询。它需要以下参数运行：

  1. --orders <path> --lineitem <path> --output <path>

可以生成订单和lineitem文件使用tpc - h基准测试套件的数据生成器工具(DBGEN)。采取以下步骤生成任意大Flink提供程序的输入文件

  1. 下载并解压DBGEN

  2. 复制makefile.suite 为 makefile，并且执行以下修改

    DATABASE = DB2
    MACHINE = LINUX
    WORKLOAD = TPCH
    CC = gcc

  3. 使用make命令构架DBGEN

  4. 使用dbgen生成lineitem和命令关系，比例因子(s)在生成的数据集1的结果大约1 GB大小。

    ./dbgen -T o -s 1

获取更多大数据资料，视频以及技术交流请加群：

watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h1NDcwNDM4MDAw_size_16_color_FFFFFF_t_70

发表评论

表情：
评论列表 （有 0 条评论，363人围观）

还没有评论，来说两句吧...

相关阅读

    相关 JDBC 批处理（Batch）

    批处理 批处理允许您将相关的SQL语句分组到一个批中，并通过对数据库的一次调用提交它们。 当您一次将多个SQL语句发送到数据库时，可以减少通信开销，从而提高性能。 >

    墨蓝墨蓝/ 0 赞/ 204 阅读