scala—如何在ApacheFlink中实现窗口函数？

hmmo2u0o 于 2021-06-05 发布在 Kafka

关注(0)|答案(1)|浏览(421)

各位，我有一个Kafka主题的来源，我分组一分钟的窗口。我想在那个窗口中做的是用window函数创建新的列，比如我想使用的sql
（除以）的和
计数（用户）超过（分区）
上的行数（）
我可以为这些操作使用数据流函数吗？或者
如何操作kafka数据将其转换为datatable并使用sqlquery？
目的地是Kafka的另一个主题。

val stream = senv
      .addSource(new FlinkKafkaConsumer[String]("flink", new SimpleStringSchema(), properties))

我试过这么做

val tableA = tableEnv.fromDataStream(stream, 'user, 'product, 'amount)

但是我得到了下面的错误

Exception in thread "main" org.apache.flink.table.api.ValidationException: Too many fields referenced from an atomic type.

试验数据

1,"beer",3
1,"beer",1
2,"beer",3
3,"diaper",4
4,"diaper",1
5,"diaper",5
6,"rubber",2

查询示例

SELECT
     user, product, amount,
     COUNT(user) OVER(PARTITION BY product) AS count_product
   FROM table;

预期业绩

1,"beer",3,3
1,"beer",1,3
2,"beer",3,3
3,"diaper",4,3
4,"diaper",1,3
5,"diaper",5,3
6,"rubber",2,1

scala streaming apache-kafka apache-flink

来源：https://stackoverflow.com/questions/61784212/how-to-implement-window-function-in-apache-flink

1条答案

按热度按时间

anhgbhbe1#

您需要将字符串解析为字段，然后重命名它们。

val env = StreamExecutionEnvironment.getExecutionEnvironment
val tEnv = StreamTableEnvironment.create(env)

val stream = env.fromElements("1,beer,3",
 "1,beer,1","2,beer,3","3,diaper,4","4,diaper,1","5,diaper,5","6,rubber,2");

val parsed = stream.map(x=> {
 val arr = x.split(",")
 (arr(0).toInt, arr(1), arr(2).toInt)
})

val tableA = tEnv.fromDataStream(parsed, $"_1" as "user", $"_2" as "product", $"_3" as "amount")

// example query
val result = tEnv.sqlQuery(s"SELECT user, product, amount from $tableA")

val rs = result.toAppendStream[(Int, String, Int)]

rs.print()

我不确定如何在flinksql中实现所需的窗口函数。或者，它可以用简单的flink实现，如下所示：

parsed.keyBy(x => x._2) // key by product id.
      .window(TumblingEventTimeWindows.of(Time.milliseconds(2)))
      .process(new ProcessWindowFunction[
        (Int, String, Int), (Int, String, Int, Int), String, TimeWindow
      ]() {
        override def process(key: String, context: Context,
                             elements: Iterable[(Int, String, Int)],
                             out: Collector[(Int, String, Int, Int)]): Unit = {
          val lst = elements.toList
          lst.foreach(x => out.collect((x._1, x._2, x._3, lst.size)))
        }
      })
      .print()

赞(0）回复(0）举报 2021-06-06

我来回答

scala—如何在ApacheFlink中实现窗口函数？

1条答案

相关问题

热门标签

最新问答