flink连kafka的json格式怎么执行SQL

[mw_shl_code=scala,true]tableEnv.connect(new Kafka().topic(topic)
                        .properties(properties)
                        .sinkPartitionerRoundRobin())

        .withFormat(new Json().failOnMissingField(false).deriveSchema())[/mw_shl_code].deriveSchema()在1.10版本中已经[backcolor=rgba(200, 200, 200, 0.1)]不推荐使用
请问一下这个方法是做什么用的。不要可不可以

deriveSchema是json format的格式的一种。
可以使用schema或者jsonSchema或者deriveSchema来定义json format

对于json是这样的：

.withFormat(
  new Json()
    .failOnMissingField(true)   // optional: flag whether to fail if a field is missing or not, false by default

    // optional: define the schema explicitly using type information. This overrides default
    // behavior that uses table's schema as format schema.
    .schema(Type.ROW(...))

    // or by using a JSON schema which parses to DECIMAL and TIMESTAMP. This also overrides default behavior.
    .jsonSchema(
      "{" +
      "  type: 'object'," +
      "  properties: {" +
      "    lon: {" +
      "      type: 'number'" +
      "    }," +
      "    rideTime: {" +
      "      type: 'string'," +
      "      format: 'date-time'" +
      "    }" +
      "  }" +
      "}"
    )
)

可参考官网
https://ci.apache.org/projects/f ... ct.html#json-format

学习看了