Write Flink SQL table definitions with Kafka source and Iceberg sink using the Table API for a streaming ETL job

Steps

1. Create a StreamTableEnvironment from a StreamExecutionEnvironment and configure the catalog to point to an Iceberg catalog (REST, Hive, or Hadoop)
2. Define the Kafka source table with CREATE TABLE using the 'kafka' connector, specifying topic, bootstrap.servers, format (json or avro), and scan.startup.mode
3. Define the Iceberg sink table with CREATE TABLE using the 'iceberg' connector and properties for catalog-name, catalog-type, database, and table
4. Write a streaming INSERT INTO iceberg_table SELECT ... FROM kafka_table with optional transformations, filtering, or aggregations in the SQL body
5. Call tableEnv.executeSql() to submit the streaming job; Flink compiles the SQL plan and runs it as a continuous DataStream job
6. Configure checkpointing on the underlying StreamExecutionEnvironment before creating the TableEnvironment to ensure Iceberg sink commits are transactional