重要|Flink SQL与kafka整合的那些事儿

1.直接注册为表// register the DataStream as Table "myTable" with fields "f0", "f1"tableEnv.registerDataStream("myTable", stream);
// register the DataStream as table "myTable2" with fields "myLong", "myString"tableEnv.registerDataStream("myTable2", stream, "myLong, myString");
2.转换为tableDataStream<Tuple2<Long, String>> stream = ...
// Convert the DataStream into a Table with default fields "f0", "f1"Table table1 = tableEnv.fromDataStream(stream);
// Convert the DataStream into a Table with fields "myLong", "myString"Table table2 = tableEnv.fromDataStream(stream, "myLong, myString");

ExternalCatalog catalog = new InMemoryExternalCatalog();
// register the ExternalCatalog catalogtableEnv.registerExternalCatalog("InMemCatalog", catalog);

import org.apache.flink.api.common.typeinfo.Types;import org.apache.flink.streaming.api.TimeCharacteristic;import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;import org.apache.flink.table.api.TableEnvironment;import org.apache.flink.table.api.java.StreamTableEnvironment;import org.apache.flink.table.descriptors.Json;import org.apache.flink.table.descriptors.Kafka;import org.apache.flink.table.descriptors.Rowtime;import org.apache.flink.table.descriptors.Schema;public class kafka2kafka {    public static void main(String[] args) throws Exception {        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);        env.setParallelism(1);        StreamTableEnvironment tEnv = TableEnvironment.getTableEnvironment(env);        tEnv.connect(                new Kafka()                        .version("0.10")                        //   "0.8", "0.9", "0.10", "0.11", and "universal"                        .topic("jsontest")                        .property("bootstrap.servers", "mt-mdh.local:9093")                        .property("group.id","test")                        .startFromLatest()        )                .withFormat(                        new Json()                                .failOnMissingField(false)                                .deriveSchema()                )                .withSchema(                        new Schema()                                .field("rowtime",Types.SQL_TIMESTAMP)                                .rowtime(new Rowtime()                                        .timestampsFromField("eventtime")                                        .watermarksPeriodicBounded(2000)                                )                                .field("fruit", Types.STRING)                                .field("number", Types.INT)                )                .inAppendMode()                .registerTableSource("source");        tEnv.connect(                new Kafka()                        .version("0.10")                        //   "0.8", "0.9", "0.10", "0.11", and "universal"                        .topic("test")                        .property("acks", "all")                        .property("retries", "0")                        .property("batch.size", "16384")                        .property("linger.ms", "10")                        .property("bootstrap.servers", "mt-mdh.local:9093")                        .sinkPartitionerFixed()        ).inAppendMode()                .withFormat(                        new Json().deriveSchema()                )                .withSchema(                        new Schema()                                .field("fruit", Types.STRING)                                .field("total", Types.INT)                                .field("time", Types.SQL_TIMESTAMP)                )                .registerTableSink("sink");        tEnv.sqlUpdate("insert into sink select fruit,sum(number),TUMBLE_END(rowtime, INTERVAL '5' SECOND) from source group by fruit,TUMBLE(rowtime, INTERVAL '5' SECOND)");        env.execute();    }}

.connect(  new Kafka()    .version("0.11")    // required: valid connector versions are                        //   "0.8", "0.9", "0.10", "0.11", and "universal"    .topic("...")       // required: topic name from which the table is read
    // optional: connector specific properties    .property("zookeeper.connect", "localhost:2181")    .property("bootstrap.servers", "localhost:9092")    .property("group.id", "testGroup")
    // optional: select a startup mode for Kafka offsets    .startFromEarliest()    .startFromLatest()    .startFromSpecificOffsets(...)
    // optional: output partitioning from Flink's partitions into Kafka's partitions    .sinkPartitionerFixed()         // each Flink partition ends up in at-most one Kafka partition (default)    .sinkPartitionerRoundRobin()    // a Flink partition is distributed to Kafka partitions round-robin    .sinkPartitionerCustom(MyCustom.class)    // use a custom FlinkKafkaPartitioner subclass)

.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())

.withSchema(  new Schema()    .field("MyField1", Types.SQL_TIMESTAMP)      .proctime()      // optional: declares this field as a processing-time attribute    .field("MyField2", Types.SQL_TIMESTAMP)      .rowtime(...)    // optional: declares this field as a event-time attribute    .field("MyField3", Types.BOOLEAN)      .from("mf3")     // optional: original field in the input that is referenced/aliased by this field)

.connect(...)
  .inAppendMode()    // otherwise: inUpsertMode() or inRetractMode()

// Converts an existing LONG or SQL_TIMESTAMP field in the input into the rowtime attribute..rowtime(  new Rowtime()    .timestampsFromField("ts_field")    // required: original field name in the input)
// Converts the assigned timestamps from a DataStream API record into the rowtime attribute// and thus preserves the assigned timestamps from the source.// This requires a source that assigns timestamps (e.g., Kafka 0.10+)..rowtime(  new Rowtime()    .timestampsFromSource())
// Sets a custom timestamp extractor to be used for the rowtime attribute.// The extractor must extend `org.apache.flink.table.sources.tsextractors.TimestampExtractor`..rowtime(  new Rowtime()    .timestampsFromExtractor(...))

// Sets a watermark strategy for ascending rowtime attributes. Emits a watermark of the maximum// observed timestamp so far minus 1. Rows that have a timestamp equal to the max timestamp// are not late..rowtime(  new Rowtime()    .watermarksPeriodicAscending())
// Sets a built-in watermark strategy for rowtime attributes which are out-of-order by a bounded time interval.// Emits watermarks which are the maximum observed timestamp minus the specified delay..rowtime(  new Rowtime()    .watermarksPeriodicBounded(2000)    // delay in milliseconds)
// Sets a built-in watermark strategy which indicates the watermarks should be preserved from the// underlying DataStream API and thus preserves the assigned watermarks from the source..rowtime(  new Rowtime()    .watermarksFromSource())