Flink原理实战每日一篇09

「爱情、让人受尽委屈。」 2021-12-23 07:05 290阅读 0赞

序：今天开始Flink SQL 跟 Flink table了，我个人觉得新手的话更要好好学习这一块的东西，Flink sql的API还是很全的，目前市面已知的很多公司已经用Flink SQL开发了很多实际业务场景，不关是Flink，几乎所有的流处理，都是向SQL靠拢，这个是主流。而且Flink的 SQL我觉得还是很强大的，值得学习的。

一，Flink Table

首先我们要知道，Flink SQL是基于 apache Calcite框架实现了SQL的标准协议，是构建在Table API之上的更高级接口。

所以这里的Table我们就简单的写一写，知道Table跟Stream的转换就好了，至于Table跟DataSet的转换啥的，有兴趣的自己去查官网的API

,1，基本的API：

import org.apache.flink.streaming.api.scala._
    import org.apache.flink.table.api.TableEnvironment
    import org.apache.flink.table.api.scala._
    
    object Flink_table {
      def main(args: Array[String]): Unit = {
        val env = StreamExecutionEnvironment.getExecutionEnvironment;
        // 创建table对象
        val tableEnv = TableEnvironment.getTableEnvironment(env)
    
        //Stream 或者 dataSet 与Table的转换
        val dataStream = env.fromElements((1, 2), (12, 23), (15, 22))
    
        // 注册成表
        tableEnv.registerDataStream("table1", dataStream)
    
        // 注册成表，指定字段名称
        tableEnv.registerDataStream("table2", dataStream, 'myLong, 'myString)
    
        // stream 转出table
        val table3 = tableEnv.fromDataStream(dataStream)
        val table4 = tableEnv.fromDataStream(dataStream, 'field1, 'field2)
    
        // Table 转成DataStream
        
    
      }
    }

2，Table 转成 DataStream

在流式计算中 Table的数据是不断的动态更新的，将Table转成Stream的话需要设定数据输出的格式：

1）Append Model模式

val stream2: DataStream\[Row\] =tableEnv.toAppendStream\[Row\](table3) /

/ 指定类型

val stream3: DataStream\[(Int, Int)\] =tableEnv.toAppendStream\[(Int,Int)\](table3)

2）Retract Model模式----这是一种更高级模式（实际使用这个比较常见）

val stream4: DataStream\[(Boolean, Row)\] =tableEnv.toRetractStream\[Row\](table3) val stream5: DataStream\[(Boolean, (Int, Int))\] =tableEnv.toRetractStream\[(Int,Int)\](table3)

注意： 这里的boolean类型代表数据更新更新，True对应的是INSERT操作更新的数据，False对应的是DELETE操作更新的数据。

3） 字段名称映射

tableEnv.fromDataStream(dataStream,'_2,'_1)
    tableEnv.fromDataStream(dataStream,'_2 as 'field1,'_1 as 'field2)

//实体类转换操作：

val dstream11: DataStream[Event] = env.fromElements(Event("a"),Event("b"))
    tableEnv.fromDataStream(dstream11,'filed_name)
    tableEnv.fromDataStream(dstream11,'name as 'name_new)

2，外部连接器：

标准格式为:

**tableEnv.connect(.........)**

**.withFormat(........)**

**.withSchema(.........)**

**.inAppendMode(.........)**

**.registerTableSource("mytable")**

1）File System Connector  本地文件连接器

// 本地文件系统连接器
    tableEnv.connect(new FileSystem().path("file:///path/temp"))

2)kafka 连接器

tableEnv.connect(
      new Kafka()
        .version("0.10") //支持 0.8 , 0.9 , 0.10 , 0.11
        .topic("test")
        .property("zookeeper,connect", "node1:2181")
        .property("bootstrap.servers", "node1:9092")
        .startFromEarliest() //最早的偏移量
        .startFromLatest() //最新的偏移量消费
        .startFromSpecificOffsets(map) //指定偏移量
        //指定分区策略
        .sinkPartitionerFixed()//每个flink分区最多被分配到一个kafka分区
        .sinkPartitionerRoundRobin() //随机分配
        .sinkPartitionerCustom(CustomPartitioner.class) //自定义分区

注意：Flink -kafka 默认情况下支持 at-least-once ，也支持 exactly-once级别的一致性保障。

3，Table format

为了传输不同的格式使用

1）CSV Format

.withFormat(
      new Csv()
        .field("field1",Types.STRING)
        .field("field2",Types.STRING)
        .fieldDelimiter(",") //指定切割符号
        .lineDelimiter("\n") //指定行切割符号
        .quoteCharacter('"')//指定字符串中的单个字符
        .commentPrefix("#") //指定comment前缀
        .ignoreFirstLine() //是否忽略第一行
        .ignoreParseErrors()//是否忽略解析错误的数据

2) JSON Format

.withFormat(
        new Json()
          .failOnMissingField(true)   // optional: flag whether to fail if a field is missing or not, false by default
    
          // required: define the schema either by using type information which parses numbers to corresponding types
          .schema(Type.ROW(...))
    
    // or by using a JSON schema which parses to DECIMAL and TIMESTAMP
    .jsonSchema(
      "{" +
        "  type: 'object'," +
        "  properties: {" +
        "    lon: {" +
        "      type: 'number'" +
        "    }," +
        "    rideTime: {" +
        "      type: 'string'," +
        "      format: 'date-time'" +
        "    }" +
        "  }" +
        "}"
    )
    
      // or use the table's schema
      .deriveSchema()

３）Ａｖｒｏ

.withFormat(
        new Avro()
          // required: define the schema either by using an Avro specific record class
          .recordClass(User.class)
    
    // or by using an Avro schema
    .avroSchema(
      "{" +
        "  \"type\": \"record\"," +
        "  \"name\": \"test\"," +
        "  \"fields\" : [" +
        "    {\"name\": \"a\", \"type\": \"long\"}," +
        "    {\"name\": \"b\", \"type\": \"string\"}" +
        "  ]" +
        "}"
    )
    )

2)Table Schema　定义了Flink Table的数据表结构

.withSchema(
      new Schema().field("id",Types.STRING)
        .proctime()   //获取process time属性
      .field("name",Types.STRING)
        .rowtime(.....) //获取event time属性
    )

注意：还可以直接接入Watermark的生成逻辑，

5 Update Modes

tableEnv.connect(........)
    .inAppendMode()  //仅仅是insert
    .inRetractMode()//insert update delete 操作
    .inUpsertMode() //insert delete 操作

6,最后写一个完整的实例

import org.apache.flink.streaming.api.scala.{StreamExecutionEnvironment, _}
    import org.apache.flink.table.api.TableEnvironment
    import org.apache.flink.table.descriptors.{Json, Kafka, Schema}
    import org.apache.flink.types.Row
    
    /**
      * Table Connector 一个案完整案例 ，从kafka读取数据指定格式直接转成了Table
      */
    object TableConnector {
      def main(args: Array[String]): Unit = {
    
        val env = StreamExecutionEnvironment.getExecutionEnvironment
        // create a TableEnvironment for streaming queries
        val tableEnv = TableEnvironment.getTableEnvironment(env)
    
        tableEnv
          .connect(
            new Kafka()
              .version("0.10")
              .topic("SM_USER_PROFILE")
              .startFromEarliest()
              .property("zookeeper.connect", "localhost:2181")
              .property("bootstrap.servers", "localhost:9092"))
          .withFormat(
            //todo 这里会报错，需要导入包flink-json-1.6.1-sql-jar
            new Json()
              .deriveSchema()
          )
          .withSchema(
            new Schema()
              .field("COD_USERNO","string")
              .field("COD_USER_ID","string")
          )
          .inAppendMode()
          //todo 注册成table
          .registerTableSource("sm_user")
    
        val stream = tableEnv.scan("sm_user")
    
      //结果转成stream 打印
        tableEnv.toAppendStream[Row](stream).print().setParallelism(1)
        env.execute("example")
    
    
      }
    }