Flink: Blink分支 1.5.1  

https://github.com/apache/flink/tree/blink

Maven Dependency:

<dependency>
  <groupId>com.alibaba.blink</groupId>
  <artifactId>flink-scala_2.11</artifactId>
  <version>1.5.1</version>
</dependency>
<dependency>
  <groupId>com.alibaba.blink</groupId>
  <artifactId>flink-streaming-scala_2.11</artifactId>
  <version>1.5.1</version>
</dependency>
<dependency>
  <groupId>com.alibaba.blink</groupId>
  <artifactId>flink-table_2.11</artifactId>
  <version>1.5.1</version>
</dependency>
<dependency>
  <groupId>com.alibaba.blink</groupId>
  <artifactId>flink-table-common</artifactId>
  <version>1.5.1</version>
</dependency>
<dependency>
  <groupId>com.alibaba.blink</groupId>
  <artifactId>flink-json</artifactId>
  <version>1.5.1</version>
</dependency>
<dependency>
  <groupId>com.alibaba.blink</groupId>
  <artifactId>flink-connector-kafka-0.10_2.11</artifactId>
  <version>1.5.1</version>
</dependency>

Kafka数据类型为Json,格式如下：

{
    "schema": {
        "type": "struct",
        "fields": [{
            "type": "struct",
            "fields": [{
                "type": "int32",
                "optional": false,
                "field": "id"
            }, {
                "type": "int32",
                "optional": true,
                "field": "age"
            }],
            "optional": true,
            "name": "postgres.public.test.Value",
            "field": "before"
        }, {
            "type": "struct",
            "fields": [{
                "type": "int32",
                "optional": false,
                "field": "id"
            }, {
                "type": "int32",
                "optional": true,
                "field": "age"
            }],
            "optional": true,
            "name": "postgres.public.test.Value",
            "field": "after"
        }, {
            "type": "struct",
            "fields": [{
                "type": "string",
                "optional": true,
                "field": "version"
            }, {
                "type": "string",
                "optional": false,
                "field": "name"
            }, {
                "type": "string",
                "optional": false,
                "field": "db"
            }, {
                "type": "int64",
                "optional": true,
                "field": "ts_usec"
            }, {
                "type": "int64",
                "optional": true,
                "field": "txId"
            }, {
                "type": "int64",
                "optional": true,
                "field": "lsn"
            }, {
                "type": "string",
                "optional": true,
                "field": "schema"
            }, {
                "type": "string",
                "optional": true,
                "field": "table"
            }, {
                "type": "boolean",
                "optional": true,
                "default": false,
                "field": "snapshot"
            }, {
                "type": "boolean",
                "optional": true,
                "field": "last_snapshot_record"
            }],
            "optional": false,
            "name": "io.debezium.connector.postgresql.Source",
            "field": "source"
        }, {
            "type": "string",
            "optional": false,
            "field": "op"
        }, {
            "type": "int64",
            "optional": true,
            "field": "ts_ms"
        }],
        "optional": false,
        "name": "postgres.public.test.Envelope"
    },
    "payload": {
        "before": null,
        "after": {
            "id": 452609,
            "age": 3
        },
        "source": {
            "version": "0.8.3.Final",
            "name": "postgres",
            "db": "postgres",
            "ts_usec": 1558601374145332000,
            "txId": 21371566,
            "lsn": 418263044043,
            "schema": "public",
            "table": "test",
            "snapshot": false,
            "last_snapshot_record": null
        },
        "op": "c",
        "ts_ms": 1558601374138
    }
}

示例代码：

import org.apache.flink.api.common.typeinfo.TypeInformation

import org.apache.flink.api.java.typeutils.ObjectArrayTypeInfo

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

import org.apache.flink.table.api.{TableEnvironment, Types}

import org.apache.flink.table.descriptors.{Json, Kafka, Schema}

import org.apache.flink.types.Row

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

 

object KafkaConsumerDemo {

  def main(args: Array[String]): Unit = {

    val env = StreamExecutionEnvironment.getExecutionEnvironment

    val tblEnv = TableEnvironment.getTableEnvironment(env)

    tblEnv.connect(new Kafka()

      .version("0.10")

      .topic("postgres.public.test")

      .property("zookeeper.connect", "BigData-Dev-1:2181")

      .property("bootstrap.servers","BigData-Dev-1:9092")

      .startFromLatest()

    ).withFormat(new Json().deriveSchema())

      .withSchema(new Schema().field("schema",Types.ROW(

        Array("type","fields","optional","name"),

        Array[TypeInformation[_]](Types.STRING,ObjectArrayTypeInfo.getInfoFor(new Array[Row](0).getClass,

          Types.ROW(Array("type","fields","optional","name","field"),Array[TypeInformation[_]](

            Types.STRING,

            ObjectArrayTypeInfo.getInfoFor(new Array[Row](0).getClass,Types.ROW(Array("type","optional","field"),Array[TypeInformation[_]](Types.STRING,Types.STRING,Types.STRING)))

            ,Types.STRING,Types.STRING,Types.STRING)

          )),Types.STRING,Types.STRING)

      )).field("payload", Types.ROW(

        Array[String]("before", "after", "source", "op", "ts_ms"),

        Array[TypeInformation[_]](

          Types.ROW(Array[String]("id", "age"), Array[TypeInformation[_]](Types.STRING, Types.STRING)),

          Types.ROW(Array[String]("id", "age"), Array[TypeInformation[_]](Types.STRING, Types.STRING)),

          Types.ROW(Array[String]("version", "name", "db", "ts_usec", "txId", "lsn", "schema", "table", "snapshot", "last_snapshot_record"), Array[TypeInformation[_]](Types.STRING, Types.STRING, Types.STRING, Types.STRING, Types.STRING, Types.STRING, Types.STRING, Types.STRING, Types.STRING, Types.STRING)),

          Types.STRING, Types.STRING))))

      .inAppendMode()

      .registerTableSource("test")

    val tableResult = tblEnv.sqlQuery("select after.id,after.age from test where  after.id is not null")

    tblEnv.toAppendStream[Row](tableResult).print()

    env.execute()

  }

}

测试：