kafka connector ddl能不能像flinkKafkaConsumer.setStartFromTimestamp(xx)一样从指定timestamp开始消费,我看文档里只提到了earliest-offset,latest-offset,group-offsets,specific-offsets
CREATE TABLE MyUserTable ( ... ) WITH ( 'connector.type' = 'kafka',
'connector.version' = '0.11', -- required: valid connector versions are -- "0.8", "0.9", "0.10", "0.11", and "universal"
'connector.topic' = 'topic_name', -- required: topic name from which the table is read
'connector.properties.zookeeper.connect' = 'localhost:2181', -- required: specify the ZooKeeper connection string 'connector.properties.bootstrap.servers' = 'localhost:9092', -- required: specify the Kafka server connection string 'connector.properties.group.id' = 'testGroup', --optional: required in Kafka consumer, specify consumer group 'connector.startup-mode' = 'earliest-offset', -- optional: valid modes are "earliest-offset", -- "latest-offset", "group-offsets", -- or "specific-offsets"
-- optional: used in case of startup mode with specific offsets 'connector.specific-offsets' = 'partition:0,offset:42;partition:1,offset:300',
'connector.sink-partitioner' = '...', -- optional: output partitioning from Flink's partitions -- into Kafka's partitions valid are "fixed" -- (each Flink partition ends up in at most one Kafka partition), -- "round-robin" (a Flink partition is distributed to -- Kafka partitions round-robin) -- "custom" (use a custom FlinkKafkaPartitioner subclass) -- optional: used in case of sink partitioner custom 'connector.sink-partitioner-class' = 'org.mycompany.MyPartitioner', 'format.type' = '...', -- required: Kafka connector requires to specify a format, ... -- the supported formats are 'csv', 'json' and 'avro'. -- Please refer to Table Formats section for more details. ) *来自志愿者整理的flink邮件归档
版权声明:本文内容由阿里云实名注册用户自发贡献,版权归原作者所有,阿里云开发者社区不拥有其著作权,亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容,填写侵权投诉表单进行举报,一经查实,本社区将立刻删除涉嫌侵权内容。