What is the syntax to Add a Column to a table using Flink SQL

user_4782811 0
Viewed 718

given below is the create statement for the table I created using flink.

CREATE TABLE event_kafkaTable (
  columnA string,
  columnB string,
  timeofevent string,
  eventTime AS TO_TIMESTAMP(TimestampConverterUtil(timeofevent)),
 WATERMARK FOR eventTime  AS eventTime - INTERVAL '5' SECOND
) WITH (
  'connector' = 'kafka',
  'topic' = 'event_name', 
  'properties.bootstrap.servers'='127.0.0.1:9092',
  'properties.group.id' = 'action_hitGroup',
  'format'= 'json',
  'scan.startup.mode'='earliest-offset',
  'json.fail-on-missing-field'='false', 
  'json.ignore-parse-errors'='true' 
)

The table above, listens to Kafka and stores data from the topic in Kafka named event_name. Now, I want to ALTER this table, by adding a new column. Following were the ALTER commands I tried running from my flink job:

1. ALTER TABLE event_kafkaTable ADD COLUMN test6 string;
2. ALTER TABLE event_kafkaTable ADD test6 string;

Both these commands threw an Flink SQL Parser exception.

The Flink's official website, https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/table/sql/alter.html, has not listed the syntax to add or drop a column from a table. Can you please let me know, what is the syntax to add or drop a column to a table using Flink's Table API.

apache-flink
1 Answers

This is not supported yet in the (default) SQL DDL syntax, but you can use the AddColumns and DropColumns Table API methods to perform those operations.

This documentation page has examples on how to use them for each supported language.

user_133124330
Related