Flink原理实战每日一篇10

谁借莪１个温暖的怀抱￠ 2021-12-19 01:39 211阅读 0赞

继续讲Flink Table ..........................

这一篇就有点乱七八糟了，都是写案例，说实话，在实际开发中都是Flink SQL，在这里我们还是手写API 然后过了，把重心放在SQL上面。。。。

下面全是手写案例 ：

1）

object Over_window {
      def main(args: Array[String]): Unit = {
    
        val env = StreamExecutionEnvironment.getExecutionEnvironment
    
    
        //todo 创建tableEnv
        val tableEnv = TableEnvironment.getTableEnvironment(env)
        val stream = env.fromElements(("aa", 192L), ("aa", 192L),("bb", 112L))
    
        val table: Table = tableEnv.fromDataStream(stream, 'id, 'time)
    
        //todo 主要四个参数   partitionBy ,  orderBy ,  preceding (following,与preceding相反)
        table.window(Over partitionBy 'id orderBy 'rowtime preceding UNBOUNDED_RANGE as 'window)
          .select('id,'time.sum over 'window,'time.max over 'window)
    
    
        //todo Over Window 聚合操作
        val rs: Table =table.window(Over partitionBy 'id orderBy 'rowtime preceding UNBOUNDED_RANGE as 'window)
          .select('id,'id.avg over 'window,'time.max over 'window)
    
        //todo 去重
    
        val distinctRs: Table = table.window(Over partitionBy 'id orderBy 'rowtime preceding UNBOUNDED_RANGE as 'window)
          .select('id,'id.avg over 'window,'time.max over 'window)
    
        table.distinct()
      }
    }

2）

object Table_API {
      def main(args: Array[String]): Unit = {
        val env = StreamExecutionEnvironment.getExecutionEnvironment
    
    
        //todo 创建tableEnv
        val tableEnv = TableEnvironment.getTableEnvironment(env)
        val stream = env.fromElements(("aa", 192L), ("aa", 192L))
        val table: Table = tableEnv.fromDataStream(stream, 'id, 'name)
    
        tableEnv.registerDataStream("table2", stream, 'myLong, 'myString)
    
    
        //todo sql查询然后转成流
        val rs: DataStream[(Boolean, (String, Long))] = table.groupBy('id)
          .select('id, 'name.sum as 'nameSum)
          .toRetractStream[(String, Long)]
    
        rs.print()
    
        env.execute()
    
        //todo  其他操作
        val table3 = tableEnv.scan("table2")
        val table4 = tableEnv.scan("table2").select('myLong,'myString)
        val table5 = tableEnv.scan("table2").select('*)
        val table6 = tableEnv.scan("table2").as('a,'b)
    
        // 算子
        val  table7= table.filter('name%2 ===0)
    
        val table8 = table.where('id ==="aa")
    
    
      }
    }

3）

object Table_join {
      def main(args: Array[String]): Unit = {
        val env = StreamExecutionEnvironment.getExecutionEnvironment
    
    
        //todo 创建tableEnv
        val tableEnv = TableEnvironment.getTableEnvironment(env)
        val stream = env.fromElements(("aa", 192L), ("aa", 192L), ("bb", 112L))
    
        val table: Table = tableEnv.fromDataStream(stream, 'key, 'time)
        val table2: Table = tableEnv.fromDataStream(stream, 'key2, 'time2)
    
        val  joinResult = table.join(table2).where('key ==='key2)
          .select('key,'time,'time2)
    
        //todo  outer Join
        val stream2 = env.fromElements(("aa", 192L,1), ("aa", 192L,2), ("bb", 112L,3))
    
        val t1: Table = tableEnv.fromDataStream(stream2, 'id1, 'var1,'var2)
        val t2: Table = tableEnv.fromDataStream(stream2, 'id2, 'var3,'var4)
    
        val  leftOutjoinResult = t1.leftOuterJoin(t2,'id1 ==='id2)
          .select('id1,'var1,'var3)
    
        val  rightOutjoinResult = t1.rightOuterJoin(t2,'id1 ==='id2)
          .select('id1,'var1,'var3)
    
        val  fullOutjoinResult = t1.fullOuterJoin(t2,'id1 ==='id2)
          .select('id1,'var1,'var3)
    
    
        //todo  TIme-windowed Join
    
        val t3: Table = tableEnv.fromDataStream(stream2, 'id1, 'var1,'var2,'time1.rowtime)
        val t4: Table = tableEnv.fromDataStream(stream2, 'id2, 'var3,'var4,'time2.rowtime)
        t3.join(t4)
          .where('id1==='id2 && 'time1 >= 'time2-10.minute && 'time1 <= 'time2+10.minute)
          .select('id1,'var1,'var3,'time1)
    
        //todo join with table Function
    
        /*val upper =new MyUpperUDTF()  //自定义table函数
        t4.join(upper('var1) as 'aa).select('id1)*/
    
        //todo 集合操作
    
        val union = t3.union(t4)
        val unionAll = t3.unionAll(t4)
    
        // 返回交集
        val intersectAllTable =t3.intersectAll(t4)
    
        //minus
        val minus =  t3.minus(t4)
        //minus All
    
        val minusall =  t3.minusAll(t4)
    
        //in
        val left_in = t3.where('id1 in(t4))
        val  in = t3.where('id1 in("aa","bb"))
    
        //todo 排序操作
        val order = t3.orderBy('id1.asc)
        val order2 = t3.orderBy('id1.desc)
        // 返回前五条
        val order3 = t3.orderBy('id1.desc).fetch(5)
    
        //忽略前10条
    
        val order4 = t3.orderBy('id1.desc).offset(10)
    
        val order5 = t3.orderBy('id1.desc).offset(10).fetch(5)
    
        //todo 数据写入
        tableEnv.registerTable("table333",t3)
        t4.insertInto("table333")
    
      }
    }

4）

object TPCHQuery3Table {
    
      def main(args: Array[String]): Unit = {
    
        val  date = "1995-03-12".toDate
        val env = ExecutionEnvironment.getExecutionEnvironment
        val tEnv = TableEnvironment.getTableEnvironment(env)
        val lineitems = getLineitemDataSet(env)
          .toTable(tEnv, 'id, 'extdPrice, 'discount, 'shipDate)
          .filter('shipDate.toDate > date)
    
        //todo 读取csv文件转成Table
        val customers = getCustomerDataSet(env)
          .toTable(tEnv, 'id, 'mktSegment)
          .filter('mktSegment === "AUTOMOBILE")
    
        val orders = getOrdersDataSet(env)
          .toTable(tEnv, 'orderId, 'custId, 'orderDate, 'shipPrio)
          .filter('orderDate.toDate < date)
    
        val items =
          orders.join(customers)
            .where('custId === 'id)
            .select('orderId, 'orderDate, 'shipPrio)
            .join(lineitems)
            .where('orderId === 'id)
            .select(
              'orderId,
              'extdPrice * (1.0f.toExpr - 'discount) as 'revenue,
              'orderDate,
              'shipPrio)
    
        val result = items
          .groupBy('orderId, 'orderDate, 'shipPrio)
          .select('orderId, 'revenue.sum as 'revenue, 'orderDate, 'shipPrio)
          .orderBy('revenue.desc, 'orderDate.asc)
    
        // emit result
        result.writeAsCsv(outputPath, "\n", "|")
    
        // execute program
        env.execute("Scala TPCH Query 3 (Table API Expression) Example")
      }
    
      case class Lineitem(id: Long, extdPrice: Double, discount: Double, shipDate: String)
      case class Customer(id: Long, mktSegment: String)
      case class Order(orderId: Long, custId: Long, orderDate: String, shipPrio: Long)
    
    
    
      private var lineitemPath: String = _
      private var customerPath: String = _
      private var ordersPath: String = _
      private var outputPath: String = _
    
      private def parseParameters(args: Array[String]): Boolean = {
        if (args.length == 4) {
          lineitemPath = args(0)
          customerPath = args(1)
          ordersPath = args(2)
          outputPath = args(3)
          true
        } else {
          System.err.println("This program expects data from the TPC-H benchmark as input data.\n" +
            " Due to legal restrictions, we can not ship generated data.\n" +
            " You can find the TPC-H data generator at http://www.tpc.org/tpch/.\n" +
            " Usage: TPCHQuery3 <lineitem-csv path> <customer-csv path> " +
            "<orders-csv path> <result path>")
          false
        }
      }
    
      private def getLineitemDataSet(env: ExecutionEnvironment): DataSet[Lineitem] = {
        env.readCsvFile[Lineitem](
          lineitemPath,
          fieldDelimiter = "|",
          includedFields = Array(0, 5, 6, 10) )
      }
    
      private def getCustomerDataSet(env: ExecutionEnvironment): DataSet[Customer] = {
        env.readCsvFile[Customer](
          customerPath,
          fieldDelimiter = "|",
          includedFields = Array(0, 6) )
      }
    
      private def getOrdersDataSet(env: ExecutionEnvironment): DataSet[Order] = {
        env.readCsvFile[Order](
          ordersPath,
          fieldDelimiter = "|",
          includedFields = Array(0, 1, 4, 7) )
      }
    }

5）

object WordCountTable {
      def main(args: Array[String]): Unit = {
    
    
        val bEnv = ExecutionEnvironment.getExecutionEnvironment
        val tEnv = TableEnvironment.getTableEnvironment(bEnv)
    
        val input = bEnv.fromElements(WC("hello", 1), WC("hello", 1), WC("hello", 1), WC("ciao", 1))
        val expr: Table = input.toTable(tEnv)
        val result = expr
          .groupBy('word)
          .select('word, 'frequency.sum as 'frequency)
          .filter('frequency > 2)
          .toDataSet[WC]
    
        result.print()
      }
    
      // *************************************************************************
      //     USER DATA TYPES
      // *************************************************************************
    
      case class WC(word: String, frequency: Long)
    
    }

总结：肯定很多人用scala写会有引入报错，要import：

import org.apache.flink.streaming.api.scala.{StreamExecutionEnvironment, _}
    import org.apache.flink.table.api.scala._
    import org.apache.flink.table.api.{Table, TableEnvironment}