zoukankan      html  css  js  c++  java
  • flink 自定义filter

    数据

    sensor_1,1547718101,35.8
    sensor_1,1547718102,22.2
    sensor_1,1547718101,55.3
    sensor_1,1547718102,24.1
    sensor_1,1547718103,57
    sensor_1,1547718103,58
    sensor_1,1547718103,59
    sensor_6,1547718101,15.4
    sensor_7,1547718102,6.7
    sensor_10,1547718205,38.1
    

      

    代码

    import org.apache.flink.api.common.functions.{FilterFunction, ReduceFunction, RichMapFunction}
    import org.apache.flink.configuration.Configuration
    import org.apache.flink.streaming.api.scala._
    
    object TransformTest {
      def main(args: Array[String]): Unit = {
        val env = StreamExecutionEnvironment.getExecutionEnvironment
        env.setParallelism(1)
        //0.从文件中读取数据
        val inputPath = "D:\ideaDemo\maven_flink\src\main\resources\sensor.txt";
        val inputStream = env.readTextFile(inputPath)
    
        //1.先转换成样例类类型(简单转换操作)
        val dataStream = inputStream.map(data => {
          val arr = data.split(",")
          SensorReding(arr(0), arr(1).toLong, arr(2).toDouble)
        })
         .filter(new MyFilter)
        env.execute()
      }
      //自定义一个函数类
      class MyFilter extends FilterFunction[SensorReding]{
        override def filter(t: SensorReding): Boolean = {
          t.id.startsWith("sensor")
        }
      }
    author@nohert
  • 相关阅读:
    LeetCode 260
    LeetCode 258
    LeetCode 237
    LeetCode 226
    LeetCode 203
    LeetCode 202
    codeforces 7D
    codefroces 7C
    codeforces 7B
    codeforces 6E (非原创)
  • 原文地址:https://www.cnblogs.com/gzgBlog/p/14928211.html
Copyright © 2011-2022 走看看