使用 Flink 解救多线程 Scala 应用

要想应用能处理一天的大流量 api 的 traces, 自然直接的想法当然是找一个 MapReduce 的 Cluster 将一天的数据 map 成多份来进行处理, 最后 reduce 到一起来算 pXX.

 1: lazy val env = ExecutionEnvironment.getExecutionEnvironment // <= 1
 2: 
 3: lazy val rawSources = (ZipkinClient.datasources foldMap interp) unsafeRunSync ()
 4: 
 5: lazy val datasources =
 6:   env.fromCollection(rawSources.toList) // <= 2
 7: 
 8: lazy val program: DataSet[Vector[String]] = datasources
 9:   .faltMap(splitByHour _)        // <= 3
10:   .rebalance()                   // <= 4
11:   .map(source =>
12:     (ZipkinClient.fetchTraces(source) foldMap interp) unsafeRunSync ()) // <= 5
13:   .groupBy(_._1.endpoint.id)   // <= 6
14:   .reduce((s1, s2) => s1 |+| s2) // <= 7
15:   .map((traces: (DataSource, Vector[Trace])) =>
16:     (ZipkinClient.aggregate(traces) foldMap interp) unsafeRunSync ())  // <= 8
17:   .map((metrics: (Vector[LatencyMetrics], DataSource)) =>
18:     (CollectorService
19:       .writeMetrics(metrics._1, Vector(metrics._2)) foldMap interp) unsafeRunSync ())
20: 

1. , DataSet是 flink batch job 的数据结果, DataSet 中的数据意味着可以并行跑在 cluster 中
2. , 类似大数据的 helloworld 计算字符个数例子, 你得先把句子拆成字符
3. , 类似于把同样的字符都归到一组, 方便下来计算

有意思的是, 你是可以本地直接 sbt run 的, 此时 flink 的 cluster 会跑在 jvm 中.

但是 跑cluster 在单个 jvm 中会耗费大量资源, 所以最佳实践其实是根据运行环境, 选择不同的 ExecutionEnvironment

lazy val env = envOrNone("EXECUTION_IN_LOCAL")
  .map(_ => ExecutionEnvironment.createCollectionsEnvironment)
  .getOrElse(ExecutionEnvironment.getExecutionEnvironment)

若是在本地运行, 比如跑功能测试或集成测试, 使用 CollectionsEnvironment 会省不少资源