DELETE 문
DELETE 문 (DELETE Statements)
DELETE 문은 제공된 필터(필터가 있는 경우)에 따라 대상 테이블에서 행 수준(row-level) 삭제를 수행하는 데 사용돼요.
출처: 문서
본문
주의: 현재
DELETE문은 배치(batch) 모드만 지원하며, 대상 테이블 커넥터가 행 수준 삭제를 지원하기 위해 SupportsRowLevelDelete 인터페이스를 구현해야 해요. 관련 인터페이스를 구현하지 않은 테이블에서DELETE를 시도하면 예외가 던져져요. 현재 Flink가 유지 관리하는 커넥터 중DELETE를 지원하는 기존 커넥터는 아직 없어요.
DELETE 문 실행하기 (Run a DELETE statement)
DELETE 문은 TableEnvironment의 executeSql() 메서드로 실행할 수 있어요. executeSql()은 즉시 Flink 작업을 제출하고, 제출된 작업과 연관된 TableResult 인스턴스를 반환해요. Python에서는 execute_sql(), SQL CLI에서는 SQL CLI에서 실행할 수 있어요.
EnvironmentSettings settings = EnvironmentSettings.newInstance().inBatchMode().build();
TableEnvironment tEnv = TableEnvironment.create(settings);
// register a table named "Orders"
tEnv.executeSql("CREATE TABLE Orders (`user` STRING, product STRING, amount INT) WITH (...)");
// insert values
tEnv.executeSql("INSERT INTO Orders VALUES ('Lili', 'Apple', 1), ('Jessica', 'Banana', 2), ('Mr.White', 'Chicken', 3)").await();
tEnv.executeSql("SELECT * FROM Orders").print();
// +--------------------------------+--------------------------------+-------------+
// | user | product | amount |
// +--------------------------------+--------------------------------+-------------+
// | Lili | Apple | 1 |
// | Jessica | Banana | 2 |
// | Mr.White | Chicken | 3 |
// +--------------------------------+--------------------------------+-------------+
// 3 rows in set
// delete by filter
tEnv.executeSql("DELETE FROM Orders WHERE `user` = 'Lili'").await();
tEnv.executeSql("SELECT * FROM Orders").print();
// +--------------------------------+--------------------------------+-------------+
// | user | product | amount |
// +--------------------------------+--------------------------------+-------------+
// | Jessica | Banana | 2 |
// | Mr.White | Chicken | 3 |
// +--------------------------------+--------------------------------+-------------+
// 2 rows in set
// delete entire table
tEnv.executeSql("DELETE FROM Orders").await();
tEnv.executeSql("SELECT * FROM Orders").print();
// Empty set
val env = StreamExecutionEnvironment.getExecutionEnvironment()
val settings = EnvironmentSettings.newInstance().inBatchMode().build()
val tEnv = StreamTableEnvironment.create(env, settings)
// register a table named "Orders"
tEnv.executeSql("CREATE TABLE Orders (`user` STRING, product STRING, amount INT) WITH (...)");
// insert values
tEnv.executeSql("INSERT INTO Orders VALUES ('Lili', 'Apple', 1), ('Jessica', 'Banana', 2), ('Mr.White', 'Chicken', 3)").await();
tEnv.executeSql("SELECT * FROM Orders").print();
// +--------------------------------+--------------------------------+-------------+
// | user | product | amount |
// +--------------------------------+--------------------------------+-------------+
// | Lili | Apple | 1 |
// | Jessica | Banana | 2 |
// | Mr.White | Chicken | 3 |
// +--------------------------------+--------------------------------+-------------+
// 3 rows in set
// delete by filter
tEnv.executeSql("DELETE FROM Orders WHERE `user` = 'Lili'").await();
tEnv.executeSql("SELECT * FROM Orders").print();
// +--------------------------------+--------------------------------+-------------+
// | user | product | amount |
// +--------------------------------+--------------------------------+-------------+
// | Jessica | Banana | 2 |
// | Mr.White | Chicken | 3 |
// +--------------------------------+--------------------------------+-------------+
// 2 rows in set
// delete entire table
tEnv.executeSql("DELETE FROM Orders").await();
tEnv.executeSql("SELECT * FROM Orders").print();
// Empty set
env_settings = EnvironmentSettings.in_batch_mode()
table_env = TableEnvironment.create(env_settings)
# register a table named "Orders"
table_env.executeSql("CREATE TABLE Orders (`user` STRING, product STRING, amount INT) WITH (...)");
# insert values
table_env.executeSql("INSERT INTO Orders VALUES ('Lili', 'Apple', 1), ('Jessica', 'Banana', 2), ('Mr.White', 'Chicken', 3)").wait();
table_env.executeSql("SELECT * FROM Orders").print();
# +--------------------------------+--------------------------------+-------------+
# | user | product | amount |
# +--------------------------------+--------------------------------+-------------+
# | Lili | Apple | 1 |
# | Jessica | Banana | 2 |
# | Mr.White | Chicken | 3 |
# +--------------------------------+--------------------------------+-------------+
# 3 rows in set
# delete by filter
table_env.executeSql("DELETE FROM Orders WHERE `user` = 'Lili'").wait();
table_env.executeSql("SELECT * FROM Orders").print();
# +--------------------------------+--------------------------------+-------------+
# | user | product | amount |
# +--------------------------------+--------------------------------+-------------+
# | Jessica | Banana | 2 |
# | Mr.White | Chicken | 3 |
# +--------------------------------+--------------------------------+-------------+
# 2 rows in set
# delete entire table
table_env.executeSql("DELETE FROM Orders").wait();
table_env.executeSql("SELECT * FROM Orders").print();
# Empty set
Flink SQL> SET 'execution.runtime-mode' = 'batch';
[INFO] Session property has been set.
Flink SQL> CREATE TABLE Orders (`user` STRING, product STRING, amount INT) with (...);
[INFO] Execute statement succeeded.
Flink SQL> INSERT INTO Orders VALUES ('Lili', 'Apple', 1), ('Jessica', 'Banana', 1), ('Mr.White', 'Chicken', 3);
[INFO] Submitting SQL update statement to the cluster...
[INFO] SQL update statement has been successfully submitted to the cluster:
Job ID: bd2c46a7b2769d5c559abd73ecde82e9
Flink SQL> SELECT * FROM Orders;
user product amount
Lili Apple 1
Jessica Banana 2
Mr.White Chicken 3
Flink SQL> DELETE FROM Orders WHERE `user` = 'Lili';
user product amount
Jessica Banana 2
Mr.White Chicken 3
DELETE ROWS
DELETE FROM [catalog_name.][db_name.]table_name [ WHERE condition ]