使用Scala版flink根据shop_data.csv数据，统计每个每个商铺的有效完成率（有效完成率=有效完成单数/接单总数）

假设shop_data.csv文件中包含以下列：订单编号、商铺名称、订单状态（已接单、已完成、已取消等）、下单时间等。

首先，我们需要创建Flink流处理环境：

import org.apache.flink.streaming.api.scala._

val env = StreamExecutionEnvironment.getExecutionEnvironment

然后，读取CSV文件并将其转换为DataStream：

val dataStream = env.readTextFile("path/to/shop_data.csv")
  .map(line => {
    val fields = line.split(",")
    val shopName = fields(1)
    val orderStatus = fields(2)
    (shopName, orderStatus)
  })

接下来，我们需要使用keyBy操作将数据流按商铺名称进行分组，并使用window操作将数据流划分为窗口：

val windowedStream = dataStream
  .keyBy(_._1)
  .window(TumblingEventTimeWindows.of(Time.minutes(10))) // 每10分钟统计一次

然后，我们可以使用reduce操作计算每个商铺的接单总数和有效完成单数：

val resultStream = windowedStream.reduce((a, b) => {
  val shopName = a._1
  val totalOrders = a._2.size + b._2.size
  val completedOrders = a._2.count(_ == "已完成") + b._2.count(_ == "已完成")
  (shopName, totalOrders, completedOrders)
})

最后，计算每个商铺的有效完成率并输出结果：

val output = resultStream.map(data => {
  val shopName = data._1
  val totalOrders = data._2
  val completedOrders = data._3
  val completionRate = completedOrders.toDouble / totalOrders.toDouble
  s"$shopName 的有效完成率为：${completionRate * 100}%"
})

output.print()

完整代码如下：

import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.streaming.api.scala.function.WindowFunction
import org.apache.flink.util.Collector

object ShopCompletionRate {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    val dataStream = env.readTextFile("path/to/shop_data.csv")
      .map(line => {
        val fields = line.split(",")
        val shopName = fields(1)
        val orderStatus = fields(2)
        (shopName, orderStatus)
      })

    val windowedStream = dataStream
      .keyBy(_._1)
      .window(TumblingEventTimeWindows.of(Time.minutes(10))) // 每10分钟统计一次

    val resultStream = windowedStream.reduce((a, b) => {
      val shopName = a._1
      val totalOrders = a._2.size + b._2.size
      val completedOrders = a._2.count(_ == "已完成") + b._2.count(_ == "已完成")
      (shopName, totalOrders, completedOrders)
    })

    val output = resultStream.map(data => {
      val shopName = data._1
      val totalOrders = data._2
      val completedOrders = data._3
      val completionRate = completedOrders.toDouble / totalOrders.toDouble
      s"$shopName 的有效完成率为：${completionRate * 100}%"
    })

    output.print()

    env.execute("ShopCompletionRate")
  }
}