diff --git a/Makefile b/Makefile
index 460dcc5..73cdd80 100644
--- a/Makefile
+++ b/Makefile
@@ -57,7 +57,7 @@ compose.cdc:
 
 .PHONY: compose.stream
 compose.stream:
-	 COMPOSE_PROFILES=flink,kafka docker-compose -f docker-compose.yml -f docker-compose-cdc.yml up
+	 COMPOSE_PROFILES=kafka docker-compose -f docker-compose.yml -f docker-compose-cdc.yml up
 
 .PHONY: compose.clean
 compose.clean:
diff --git a/docker-compose.yml b/docker-compose.yml
index ac2d2b0..28600a1 100644
--- a/docker-compose.yml
+++ b/docker-compose.yml
@@ -244,7 +244,7 @@ services:
 
     working_dir: /opt/flink
     ports:
-      - "8081:8081"
+      - "8082:8081"
       - "6123:6123"
 
     environment:
diff --git a/project-flink/.gitignore b/project-flink/.gitignore
new file mode 100644
index 0000000..d18d270
--- /dev/null
+++ b/project-flink/.gitignore
@@ -0,0 +1,44 @@
+### Java ###
+# Compiled class file
+*.class
+
+# Log file
+*.log
+
+# BlueJ files
+*.ctxt
+
+# Mobile Tools for Java (J2ME)
+.mtj.tmp/
+
+# Package Files #
+*.jar
+*.war
+*.nar
+*.ear
+*.zip
+*.tar.gz
+*.rar
+
+# virtual machine crash logs, see http://www.java.com/en/download/help/error_hotspot.xml
+hs_err_pid*
+replay_pid*
+
+### Maven ###
+target/
+pom.xml.tag
+pom.xml.releaseBackup
+pom.xml.versionsBackup
+pom.xml.next
+release.properties
+dependency-reduced-pom.xml
+buildNumber.properties
+.mvn/timing.properties
+# https://github.com/takari/maven-wrapper#usage-without-binary-jar
+.mvn/wrapper/maven-wrapper.jar
+
+# Eclipse m2e generated files
+# Eclipse Core
+.project
+# JDT-specific (Eclipse Java Development Tools)
+.classpath
diff --git a/project-flink/.idea/icon.png b/project-flink/.idea/icon.png
new file mode 100644
index 0000000..37c980e
Binary files /dev/null and b/project-flink/.idea/icon.png differ
diff --git a/project-flink/flink-sql-iceberg/pom.xml b/project-flink/flink-sql-iceberg/pom.xml
new file mode 100644
index 0000000..8b241f4
--- /dev/null
+++ b/project-flink/flink-sql-iceberg/pom.xml
@@ -0,0 +1,223 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+  xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+  xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+  <modelVersion>4.0.0</modelVersion>
+
+  <parent>
+    <groupId>com.github.lambda.lakehouse</groupId>
+    <artifactId>project-flink</artifactId>
+    <version>1.0-SNAPSHOT</version>
+    <relativePath>../pom.xml</relativePath>
+  </parent>
+
+  <artifactId>flink-sql-iceberg</artifactId>
+
+  <properties>
+  </properties>
+
+  <dependencies>
+
+    <dependency>
+      <groupId>org.apache.flink</groupId>
+      <artifactId>flink-json</artifactId>
+      <version>${dep.version.flink}</version>
+      <scope>provided</scope>
+    </dependency>
+
+    <dependency>
+      <groupId>org.apache.flink</groupId>
+      <artifactId>flink-streaming-java</artifactId>
+      <version>${dep.version.flink}</version>
+      <scope>provided</scope>
+    </dependency>
+
+    <dependency>
+      <groupId>org.apache.flink</groupId>
+      <artifactId>flink-table-api-java</artifactId>
+      <version>${dep.version.flink}</version>
+      <scope>provided</scope>
+    </dependency>
+
+    <dependency>
+      <groupId>org.apache.flink</groupId>
+      <artifactId>flink-table-planner_2.12</artifactId>
+      <version>${dep.version.flink}</version>
+    </dependency>
+
+    <!-- Parquet Dependency -->
+
+    <dependency>
+      <groupId>org.apache.flink</groupId>
+      <artifactId>flink-connector-files</artifactId>
+      <version>${dep.version.flink}</version>
+      <scope>provided</scope>
+    </dependency>
+
+    <dependency>
+      <groupId>org.apache.flink</groupId>
+      <artifactId>flink-parquet</artifactId>
+      <version>${dep.version.flink}</version>
+      <scope>provided</scope>
+    </dependency>
+
+    <dependency>
+      <groupId>org.apache.flink</groupId>
+      <artifactId>flink-s3-fs-hadoop</artifactId>
+      <version>${dep.version.flink}</version>
+      <scope>provided</scope>
+    </dependency>
+
+    <!-- Kafka Dependency -->
+
+    <dependency>
+      <groupId>org.apache.flink</groupId>
+      <artifactId>flink-connector-kafka</artifactId>
+      <version>${dep.version.flink}</version>
+    </dependency>
+
+    <dependency>
+      <groupId>org.apache.flink</groupId>
+      <artifactId>flink-connector-kafka</artifactId>
+      <version>${dep.version.flink}</version>
+      <scope>provided</scope>
+    </dependency>
+
+    <dependency>
+      <groupId>org.apache.flink</groupId>
+      <artifactId>flink-sql-connector-kafka</artifactId>
+      <version>${dep.version.flink}</version>
+      <scope>provided</scope>
+    </dependency>
+
+    <dependency>
+      <groupId>org.apache.kafka</groupId>
+      <artifactId>kafka-clients</artifactId>
+      <version>${dep.version.kafka}</version>
+    </dependency>
+
+    <!-- Iceberg Dependency -->
+    <dependency>
+      <groupId>org.apache.iceberg</groupId>
+      <artifactId>iceberg-flink-runtime-${dep.version.flinkShort}</artifactId>
+      <version>${dep.version.iceberg}</version>
+    </dependency>
+
+    <dependency>
+      <groupId>org.apache.iceberg</groupId>
+      <artifactId>iceberg-hive-runtime</artifactId>
+      <version>${dep.version.iceberg}</version>
+    </dependency>
+
+    <!-- Extra Dependency -->
+    <dependency>
+      <groupId>software.amazon.awssdk</groupId>
+      <artifactId>bundle</artifactId>
+      <version>${dep.version.awssdk}</version>
+    </dependency>
+
+    <!-- Hadoop Dependency -->
+    <dependency>
+      <groupId>org.apache.hadoop</groupId>
+      <artifactId>hadoop-mapreduce-client-core</artifactId>
+      <version>${dep.version.hadoop}</version>
+      <exclusions>
+        <exclusion>
+          <groupId>com.sun.jersey</groupId>
+          <artifactId>jersey-core</artifactId>
+        </exclusion>
+      </exclusions>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.hadoop</groupId>
+      <artifactId>hadoop-mapreduce-client-common</artifactId>
+      <version>${dep.version.hadoop}</version>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.hadoop</groupId>
+      <artifactId>hadoop-mapreduce-client-jobclient</artifactId>
+      <version>${dep.version.hadoop}</version>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.hadoop</groupId>
+      <artifactId>hadoop-hdfs</artifactId>
+      <version>${dep.version.hadoop}</version>
+      <exclusions>
+        <exclusion>
+          <groupId>com.sun.jersey</groupId>
+          <artifactId>jersey-core</artifactId>
+        </exclusion>
+      </exclusions>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.hadoop</groupId>
+      <artifactId>hadoop-common</artifactId>
+      <version>${dep.version.hadoop}</version>
+      <exclusions>
+        <exclusion>
+          <groupId>com.sun.jersey</groupId>
+          <artifactId>jersey-core</artifactId>
+        </exclusion>
+      </exclusions>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.hadoop</groupId>
+      <artifactId>hadoop-aws</artifactId>
+      <version>${dep.version.hadoop}</version>
+    </dependency>
+
+    <!-- Hive Dependency -->
+    <dependency>
+      <groupId>org.apache.flink</groupId>
+      <artifactId>flink-connector-hive_2.12</artifactId>
+      <version>${dep.version.flink}</version>
+      <scope>provided</scope>
+    </dependency>
+
+    <dependency>
+      <groupId>org.apache.hive</groupId>
+      <artifactId>hive-exec</artifactId>
+      <version>${dep.version.hive}</version>
+      <exclusions>
+        <exclusion>
+          <groupId>org.apache.orc</groupId>
+          <artifactId>orc-core</artifactId>
+        </exclusion>
+      </exclusions>
+    </dependency>
+
+  </dependencies>
+
+  <build>
+    <plugins>
+      <!-- The maven-shade plugin creates a fat jar that contains all
+        dependencies. -->
+      <plugin>
+        <groupId>org.apache.maven.plugins</groupId>
+        <artifactId>maven-shade-plugin</artifactId>
+        <version>3.4.1</version>
+        <configuration>
+          <shadedArtifactAttached>true</shadedArtifactAttached>
+          <shadedClassifierName>combined</shadedClassifierName>
+        </configuration>
+        <executions>
+          <execution>
+            <phase>package</phase>
+            <goals>
+              <goal>shade</goal>
+            </goals>
+          </execution>
+        </executions>
+      </plugin>
+      <plugin>
+        <groupId>org.apache.maven.plugins</groupId>
+        <artifactId>maven-compiler-plugin</artifactId>
+        <configuration>
+          <source>8</source>
+          <target>8</target>
+        </configuration>
+      </plugin>
+    </plugins>
+  </build>
+
+</project>
diff --git a/project-flink/flink-sql-iceberg/src/main/java/com/github/lambda/lakehouse/FlinkAppKafkaToIceberg.java b/project-flink/flink-sql-iceberg/src/main/java/com/github/lambda/lakehouse/FlinkAppKafkaToIceberg.java
new file mode 100644
index 0000000..96a1718
--- /dev/null
+++ b/project-flink/flink-sql-iceberg/src/main/java/com/github/lambda/lakehouse/FlinkAppKafkaToIceberg.java
@@ -0,0 +1,98 @@
+package com.github.lambda.lakehouse;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;
+import org.apache.flink.streaming.api.CheckpointingMode;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.Table;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class FlinkAppKafkaToIceberg {
+
+  private static final Logger LOG = LoggerFactory.getLogger(FlinkAppKafkaToIceberg.class);
+
+  public static void main(String[] args) throws Exception {
+
+    TableEnvironment tableEnv = buildTableEnvironment();
+
+    Table tableRawCustomers = buildSourceTable("raw_customers", tableEnv);
+    Table tableAggrCustomers = buildSinkTable("inventory", "sink_customers", tableEnv);
+
+    tableEnv.executeSql("INSERT INTO sink_customers SELECT id, weight FROM raw_customers");
+  }
+
+  public static Table buildSinkTable(String dbName, String tableName, TableEnvironment tableEnv) {
+    String query = ""
+        + "CREATE TABLE " + dbName + "." + tableName + " (\n"
+        + "   id BIGINT,\n"
+        + "   weight DECIMAL(38, 10),\n"
+        + "   PRIMARY KEY (id) NOT ENFORCED\n"
+        + ") "
+        + "WITH (\n"
+        + " 'connector' = 'iceberg',\n"
+        + " 'catalog-name' = 'hive_prod',\n"
+        + " 'catalog-database' = '" + dbName + "',\n"
+        + " 'catalog-table' = 'customers',\n"
+        + " 'uri' = 'thrift://localhost:9083',\n"
+        + " 'warehouse' = 's3a://datalake'\n"
+        + ");\n";
+    tableEnv.executeSql(query);
+    tableEnv.executeSql("SHOW CREATE TABLE " + dbName + "." + tableName).print();
+
+    Table table = tableEnv.from(dbName + "." + tableName);
+
+    return table;
+  }
+
+  public static Table buildSourceTable(String tableName, TableEnvironment tableEnv) {
+    String query = ""
+        + "CREATE TABLE " + tableName + " (\n"
+        + "   origin_ts TIMESTAMP(3) METADATA FROM 'value.ingestion-timestamp' VIRTUAL,\n"
+        + "   event_time TIMESTAMP(3) METADATA FROM 'value.source.timestamp' VIRTUAL,\n"
+        + "   origin_database STRING METADATA FROM 'value.source.database' VIRTUAL,\n"
+        + "   origin_schema STRING METADATA FROM 'value.source.schema' VIRTUAL,\n"
+        + "   origin_table STRING METADATA FROM 'value.source.table' VIRTUAL,\n"
+        + "   origin_properties MAP<STRING, STRING> METADATA FROM 'value.source.properties' VIRTUAL,\n"
+        + "   id BIGINT,\n"
+        + "   name STRING,\n"
+        + "   description STRING,\n"
+        + "   weight DECIMAL(38, 10)\n" + ") "
+        + "WITH (\n"
+        + " 'connector' = 'kafka',\n"
+        + " 'topic' = 'cdc-json.inventory.data.inventory.customers',\n"
+        + " 'properties.bootstrap.servers' = 'localhost:9092',\n"
+        + " 'properties.group.id' = 'testGroup',\n"
+        + " 'properties.auto.offset.reset' = 'earliest',\n"
+        + " 'scan.startup.mode' = 'earliest-offset',\n"
+        + " 'format' = 'debezium-json',\n"
+        + " 'debezium-json.schema-include' = 'true',\n"
+        + " 'debezium-json.ignore-parse-errors' = 'false'\n"
+        + ");\n";
+    tableEnv.executeSql(query);
+    tableEnv.executeSql("SHOW CREATE TABLE " + tableName).print();
+
+    Table table = tableEnv.from(tableName);
+
+    return table;
+  }
+
+  public static StreamTableEnvironment buildTableEnvironment() {
+    // TODO (Kun): Handle Parameters
+    // - https://nightlies.apache.org/flink/flink-docs-master/docs/dev/datastream/application_parameters/
+    Configuration conf = new Configuration();
+    StreamExecutionEnvironment env = StreamExecutionEnvironment
+        .createLocalEnvironmentWithWebUI(conf);
+    env.getCheckpointConfig().setCheckpointInterval(10000L);
+//    env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
+//    env.setStateBackend(new EmbeddedRocksDBStateBackend());
+//    env.getCheckpointConfig().setCheckpointStorage("file:///tmp/flink-checkpoint");
+//    env.setDefaultSavepointDirectory("file:///tmp/flink-savepoint");
+
+    StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
+
+    return tableEnv;
+  }
+}
diff --git a/project-flink/flink-sql-iceberg/src/main/resources/core-site.xml b/project-flink/flink-sql-iceberg/src/main/resources/core-site.xml
new file mode 100644
index 0000000..b0dbd6e
--- /dev/null
+++ b/project-flink/flink-sql-iceberg/src/main/resources/core-site.xml
@@ -0,0 +1,34 @@
+<configuration>
+  <property>
+    <name>fs.s3a.endpoint</name>
+    <value>http://localhost:9000</value>
+  </property>
+  <property>
+    <name>fs.s3a.access.key</name>
+    <value>minio</value>
+  </property>
+  <property>
+    <name>fs.s3a.secret.key</name>
+    <value>minio123</value>
+  </property>
+  <property>
+    <name>fs.s3a.path.style.access</name>
+    <value>true</value>
+  </property>
+  <property>
+    <name>fs.s3.endpoint</name>
+    <value>http://localhost:9000</value>
+  </property>
+  <property>
+    <name>fs.s3.access.key</name>
+    <value>minio</value>
+  </property>
+  <property>
+    <name>fs.s3.secret.key</name>
+    <value>minio123</value>
+  </property>
+  <property>
+    <name>fs.s3.path.style.access</name>
+    <value>true</value>
+  </property>
+</configuration>
diff --git a/project-flink/flink-sql-iceberg/src/main/resources/log4j.properties b/project-flink/flink-sql-iceberg/src/main/resources/log4j.properties
new file mode 100644
index 0000000..1ab559e
--- /dev/null
+++ b/project-flink/flink-sql-iceberg/src/main/resources/log4j.properties
@@ -0,0 +1,5 @@
+log4j.rootLogger=INFO, console
+
+log4j.appender.console=org.apache.log4j.ConsoleAppender
+log4j.appender.console.layout=org.apache.log4j.PatternLayout
+log4j.appender.console.layout.ConversionPattern=%d{HH:mm:ss,SSS} %-5p %-60c %x - %m%n
diff --git a/project-flink/flink-sql-kafka/pom.xml b/project-flink/flink-sql-kafka/pom.xml
new file mode 100644
index 0000000..0671d7f
--- /dev/null
+++ b/project-flink/flink-sql-kafka/pom.xml
@@ -0,0 +1,208 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+  xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+  xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+  <modelVersion>4.0.0</modelVersion>
+
+  <parent>
+    <groupId>com.github.lambda.lakehouse</groupId>
+    <artifactId>project-flink</artifactId>
+    <version>1.0-SNAPSHOT</version>
+    <relativePath>../pom.xml</relativePath>
+  </parent>
+
+  <artifactId>flink-sql-kafka</artifactId>
+
+  <properties>
+  </properties>
+
+  <dependencies>
+
+    <dependency>
+      <groupId>org.apache.flink</groupId>
+      <artifactId>flink-json</artifactId>
+      <version>${dep.version.flink}</version>
+      <scope>provided</scope>
+    </dependency>
+
+    <dependency>
+      <groupId>org.apache.flink</groupId>
+      <artifactId>flink-streaming-java</artifactId>
+      <version>${dep.version.flink}</version>
+      <scope>provided</scope>
+    </dependency>
+
+    <dependency>
+      <groupId>org.apache.flink</groupId>
+      <artifactId>flink-table-api-java</artifactId>
+      <version>${dep.version.flink}</version>
+      <scope>provided</scope>
+    </dependency>
+
+    <dependency>
+      <groupId>org.apache.flink</groupId>
+      <artifactId>flink-table-planner_2.12</artifactId>
+      <version>${dep.version.flink}</version>
+    </dependency>
+
+    <!-- Kafka Dependency -->
+    <dependency>
+      <groupId>org.apache.flink</groupId>
+      <artifactId>flink-connector-kafka</artifactId>
+      <version>${dep.version.flink}</version>
+    </dependency>
+
+    <dependency>
+      <groupId>org.apache.flink</groupId>
+      <artifactId>flink-connector-kafka</artifactId>
+      <version>${dep.version.flink}</version>
+      <scope>provided</scope>
+    </dependency>
+
+    <dependency>
+      <groupId>org.apache.flink</groupId>
+      <artifactId>flink-sql-connector-kafka</artifactId>
+      <version>${dep.version.flink}</version>
+      <scope>provided</scope>
+    </dependency>
+
+    <dependency>
+      <groupId>org.apache.kafka</groupId>
+      <artifactId>kafka-clients</artifactId>
+      <version>${dep.version.kafka}</version>
+    </dependency>
+
+    <!-- Parquet Dependency -->
+    <dependency>
+      <groupId>org.apache.flink</groupId>
+      <artifactId>flink-connector-files</artifactId>
+      <version>${dep.version.flink}</version>
+      <scope>provided</scope>
+    </dependency>
+
+    <dependency>
+      <groupId>org.apache.flink</groupId>
+      <artifactId>flink-parquet</artifactId>
+      <version>${dep.version.flink}</version>
+      <scope>provided</scope>
+    </dependency>
+
+    <dependency>
+      <groupId>org.apache.flink</groupId>
+      <artifactId>flink-s3-fs-hadoop</artifactId>
+      <version>${dep.version.flink}</version>
+      <scope>provided</scope>
+    </dependency>
+
+    <!-- Extra Dependency -->
+    <dependency>
+      <groupId>software.amazon.awssdk</groupId>
+      <artifactId>bundle</artifactId>
+      <version>${dep.version.awssdk}</version>
+    </dependency>
+
+    <!-- Hadoop Dependency -->
+    <dependency>
+      <groupId>org.apache.hadoop</groupId>
+      <artifactId>hadoop-mapreduce-client-core</artifactId>
+      <version>${dep.version.hadoop}</version>
+      <exclusions>
+        <exclusion>
+          <groupId>com.sun.jersey</groupId>
+          <artifactId>jersey-core</artifactId>
+        </exclusion>
+      </exclusions>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.hadoop</groupId>
+      <artifactId>hadoop-mapreduce-client-common</artifactId>
+      <version>${dep.version.hadoop}</version>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.hadoop</groupId>
+      <artifactId>hadoop-mapreduce-client-jobclient</artifactId>
+      <version>${dep.version.hadoop}</version>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.hadoop</groupId>
+      <artifactId>hadoop-hdfs</artifactId>
+      <version>${dep.version.hadoop}</version>
+      <exclusions>
+        <exclusion>
+          <groupId>com.sun.jersey</groupId>
+          <artifactId>jersey-core</artifactId>
+        </exclusion>
+      </exclusions>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.hadoop</groupId>
+      <artifactId>hadoop-common</artifactId>
+      <version>${dep.version.hadoop}</version>
+      <exclusions>
+        <exclusion>
+          <groupId>com.sun.jersey</groupId>
+          <artifactId>jersey-core</artifactId>
+        </exclusion>
+      </exclusions>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.hadoop</groupId>
+      <artifactId>hadoop-aws</artifactId>
+      <version>${dep.version.hadoop}</version>
+    </dependency>
+
+    <!-- Hive Dependency -->
+    <dependency>
+      <groupId>org.apache.flink</groupId>
+      <artifactId>flink-connector-hive_2.12</artifactId>
+      <version>${dep.version.flink}</version>
+      <scope>provided</scope>
+    </dependency>
+
+    <dependency>
+      <groupId>org.apache.hive</groupId>
+      <artifactId>hive-exec</artifactId>
+      <version>${dep.version.hive}</version>
+      <exclusions>
+        <exclusion>
+          <groupId>org.apache.orc</groupId>
+          <artifactId>orc-core</artifactId>
+        </exclusion>
+      </exclusions>
+    </dependency>
+
+  </dependencies>
+
+  <build>
+    <plugins>
+      <!-- The maven-shade plugin creates a fat jar that contains all
+        dependencies. -->
+      <plugin>
+        <groupId>org.apache.maven.plugins</groupId>
+        <artifactId>maven-shade-plugin</artifactId>
+        <version>3.4.1</version>
+        <configuration>
+          <shadedArtifactAttached>true</shadedArtifactAttached>
+          <shadedClassifierName>combined</shadedClassifierName>
+        </configuration>
+        <executions>
+          <execution>
+            <phase>package</phase>
+            <goals>
+              <goal>shade</goal>
+            </goals>
+          </execution>
+        </executions>
+      </plugin>
+      <plugin>
+        <groupId>org.apache.maven.plugins</groupId>
+        <artifactId>maven-compiler-plugin</artifactId>
+        <configuration>
+          <source>8</source>
+          <target>8</target>
+        </configuration>
+      </plugin>
+    </plugins>
+  </build>
+
+</project>
diff --git a/project-flink/flink-sql-kafka/src/main/java/com/github/lambda/lakehouse/FlinkKafkaToKafka.java b/project-flink/flink-sql-kafka/src/main/java/com/github/lambda/lakehouse/FlinkKafkaToKafka.java
new file mode 100644
index 0000000..e5bea7c
--- /dev/null
+++ b/project-flink/flink-sql-kafka/src/main/java/com/github/lambda/lakehouse/FlinkKafkaToKafka.java
@@ -0,0 +1,98 @@
+package com.github.lambda.lakehouse;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;
+import org.apache.flink.streaming.api.CheckpointingMode;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.Table;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class FlinkKafkaToKafka {
+
+  private static final Logger LOG = LoggerFactory.getLogger(FlinkKafkaToKafka.class);
+
+  public static void main(String[] args) throws Exception {
+
+    TableEnvironment tableEnv = buildTableEnvironment();
+
+    Table tableRawCustomers = buildSourceTable("raw_customers", tableEnv);
+    Table tableAggrCustomers = buildSinkTable("aggr_customers", tableEnv);
+
+    tableEnv.executeSql("INSERT INTO aggr_customers SELECT id, weight FROM raw_customers");
+  }
+
+  public static Table buildSinkTable(String tableName, TableEnvironment tableEnv) {
+    String query = ""
+        + "CREATE TABLE " + tableName + " (\n"
+        + "   id BIGINT,\n"
+        + "   weight DECIMAL(38, 10),\n"
+        + "   PRIMARY KEY (id) NOT ENFORCED\n"
+        + ") "
+        + "WITH (\n"
+        + " 'connector' = 'upsert-kafka',\n"
+        + " 'topic' = 'aggregation.customers',\n"
+        + " 'properties.bootstrap.servers' = 'localhost:9092',\n"
+        + " 'properties.allow.auto.create.topics' = 'true',\n"
+        + " 'value.format' = 'json',\n"
+        + " 'key.format' = 'json'\n"
+        + ");\n";
+    tableEnv.executeSql(query);
+    tableEnv.executeSql("SHOW CREATE TABLE " + tableName).print();
+
+    Table table = tableEnv.from(tableName);
+
+    return table;
+  }
+
+  public static Table buildSourceTable(String tableName, TableEnvironment tableEnv) {
+    String query = ""
+        + "CREATE TABLE " + tableName + " (\n"
+        + "   origin_ts TIMESTAMP(3) METADATA FROM 'value.ingestion-timestamp' VIRTUAL,\n"
+        + "   event_time TIMESTAMP(3) METADATA FROM 'value.source.timestamp' VIRTUAL,\n"
+        + "   origin_database STRING METADATA FROM 'value.source.database' VIRTUAL,\n"
+        + "   origin_schema STRING METADATA FROM 'value.source.schema' VIRTUAL,\n"
+        + "   origin_table STRING METADATA FROM 'value.source.table' VIRTUAL,\n"
+        + "   origin_properties MAP<STRING, STRING> METADATA FROM 'value.source.properties' VIRTUAL,\n"
+        + "   id BIGINT,\n"
+        + "   name STRING,\n"
+        + "   description STRING,\n"
+        + "   weight DECIMAL(38, 10)\n" + ") "
+        + "WITH (\n"
+        + " 'connector' = 'kafka',\n"
+        + " 'topic' = 'cdc-json.inventory.data.inventory.customers',\n"
+        + " 'properties.bootstrap.servers' = 'localhost:9092',\n"
+        + " 'properties.group.id' = 'testGroup',\n"
+        + " 'properties.auto.offset.reset' = 'earliest',\n"
+        + " 'scan.startup.mode' = 'earliest-offset',\n"
+        + " 'format' = 'debezium-json',\n"
+        + " 'debezium-json.schema-include' = 'true',\n"
+        + " 'debezium-json.ignore-parse-errors' = 'false'\n"
+        + ");\n";
+    tableEnv.executeSql(query);
+    tableEnv.executeSql("SHOW CREATE TABLE " + tableName).print();
+
+    Table table = tableEnv.from(tableName);
+
+    return table;
+  }
+
+  public static StreamTableEnvironment buildTableEnvironment() {
+    // TODO (Kun): Handle Parameters
+    // - https://nightlies.apache.org/flink/flink-docs-master/docs/dev/datastream/application_parameters/
+    Configuration conf = new Configuration();
+    StreamExecutionEnvironment env = StreamExecutionEnvironment
+        .createLocalEnvironmentWithWebUI(conf);
+    env.getCheckpointConfig().setCheckpointInterval(30000L);
+    env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
+    env.setStateBackend(new EmbeddedRocksDBStateBackend());
+    env.getCheckpointConfig().setCheckpointStorage("file:///tmp/flink-checkpoint");
+    env.setDefaultSavepointDirectory("file:///tmp/flink-savepoint");
+
+    StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
+
+    return tableEnv;
+  }
+}
diff --git a/project-flink/flink-sql-kafka/src/main/resources/core-site.xml b/project-flink/flink-sql-kafka/src/main/resources/core-site.xml
new file mode 100644
index 0000000..b0dbd6e
--- /dev/null
+++ b/project-flink/flink-sql-kafka/src/main/resources/core-site.xml
@@ -0,0 +1,34 @@
+<configuration>
+  <property>
+    <name>fs.s3a.endpoint</name>
+    <value>http://localhost:9000</value>
+  </property>
+  <property>
+    <name>fs.s3a.access.key</name>
+    <value>minio</value>
+  </property>
+  <property>
+    <name>fs.s3a.secret.key</name>
+    <value>minio123</value>
+  </property>
+  <property>
+    <name>fs.s3a.path.style.access</name>
+    <value>true</value>
+  </property>
+  <property>
+    <name>fs.s3.endpoint</name>
+    <value>http://localhost:9000</value>
+  </property>
+  <property>
+    <name>fs.s3.access.key</name>
+    <value>minio</value>
+  </property>
+  <property>
+    <name>fs.s3.secret.key</name>
+    <value>minio123</value>
+  </property>
+  <property>
+    <name>fs.s3.path.style.access</name>
+    <value>true</value>
+  </property>
+</configuration>
diff --git a/project-flink/flink-sql-kafka/src/main/resources/log4j.properties b/project-flink/flink-sql-kafka/src/main/resources/log4j.properties
new file mode 100644
index 0000000..1ab559e
--- /dev/null
+++ b/project-flink/flink-sql-kafka/src/main/resources/log4j.properties
@@ -0,0 +1,5 @@
+log4j.rootLogger=INFO, console
+
+log4j.appender.console=org.apache.log4j.ConsoleAppender
+log4j.appender.console.layout=org.apache.log4j.PatternLayout
+log4j.appender.console.layout.ConversionPattern=%d{HH:mm:ss,SSS} %-5p %-60c %x - %m%n
diff --git a/project-flink/flink-sql-parquet/pom.xml b/project-flink/flink-sql-parquet/pom.xml
new file mode 100644
index 0000000..16a0821
--- /dev/null
+++ b/project-flink/flink-sql-parquet/pom.xml
@@ -0,0 +1,181 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+  xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+  xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+  <modelVersion>4.0.0</modelVersion>
+
+  <parent>
+    <groupId>com.github.lambda.lakehouse</groupId>
+    <artifactId>project-flink</artifactId>
+    <version>1.0-SNAPSHOT</version>
+    <relativePath>../pom.xml</relativePath>
+  </parent>
+
+  <artifactId>flink-sql-parquet</artifactId>
+
+  <properties>
+  </properties>
+
+  <dependencies>
+
+    <dependency>
+      <groupId>org.apache.flink</groupId>
+      <artifactId>flink-json</artifactId>
+      <version>${dep.version.flink}</version>
+      <scope>provided</scope>
+    </dependency>
+
+    <dependency>
+      <groupId>org.apache.flink</groupId>
+      <artifactId>flink-streaming-java</artifactId>
+      <version>${dep.version.flink}</version>
+      <scope>provided</scope>
+    </dependency>
+
+    <dependency>
+      <groupId>org.apache.flink</groupId>
+      <artifactId>flink-table-api-java</artifactId>
+      <version>${dep.version.flink}</version>
+      <scope>provided</scope>
+    </dependency>
+
+    <dependency>
+      <groupId>org.apache.flink</groupId>
+      <artifactId>flink-table-planner_2.12</artifactId>
+      <version>${dep.version.flink}</version>
+    </dependency>
+
+    <!-- Parquet Dependency -->
+    <dependency>
+      <groupId>org.apache.flink</groupId>
+      <artifactId>flink-connector-files</artifactId>
+      <version>${dep.version.flink}</version>
+      <scope>provided</scope>
+    </dependency>
+
+    <dependency>
+      <groupId>org.apache.flink</groupId>
+      <artifactId>flink-parquet</artifactId>
+      <version>${dep.version.flink}</version>
+      <scope>provided</scope>
+    </dependency>
+
+    <dependency>
+      <groupId>org.apache.flink</groupId>
+      <artifactId>flink-s3-fs-hadoop</artifactId>
+      <version>${dep.version.flink}</version>
+      <scope>provided</scope>
+    </dependency>
+
+    <!-- Extra Dependency -->
+    <dependency>
+      <groupId>software.amazon.awssdk</groupId>
+      <artifactId>bundle</artifactId>
+      <version>${dep.version.awssdk}</version>
+    </dependency>
+
+    <!-- Hadoop Dependency -->
+    <dependency>
+      <groupId>org.apache.hadoop</groupId>
+      <artifactId>hadoop-mapreduce-client-core</artifactId>
+      <version>${dep.version.hadoop}</version>
+      <exclusions>
+        <exclusion>
+          <groupId>com.sun.jersey</groupId>
+          <artifactId>jersey-core</artifactId>
+        </exclusion>
+      </exclusions>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.hadoop</groupId>
+      <artifactId>hadoop-mapreduce-client-common</artifactId>
+      <version>${dep.version.hadoop}</version>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.hadoop</groupId>
+      <artifactId>hadoop-mapreduce-client-jobclient</artifactId>
+      <version>${dep.version.hadoop}</version>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.hadoop</groupId>
+      <artifactId>hadoop-hdfs</artifactId>
+      <version>${dep.version.hadoop}</version>
+      <exclusions>
+        <exclusion>
+          <groupId>com.sun.jersey</groupId>
+          <artifactId>jersey-core</artifactId>
+        </exclusion>
+      </exclusions>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.hadoop</groupId>
+      <artifactId>hadoop-common</artifactId>
+      <version>${dep.version.hadoop}</version>
+      <exclusions>
+        <exclusion>
+          <groupId>com.sun.jersey</groupId>
+          <artifactId>jersey-core</artifactId>
+        </exclusion>
+      </exclusions>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.hadoop</groupId>
+      <artifactId>hadoop-aws</artifactId>
+      <version>${dep.version.hadoop}</version>
+    </dependency>
+
+    <!-- Hive Dependency -->
+    <dependency>
+      <groupId>org.apache.flink</groupId>
+      <artifactId>flink-connector-hive_2.12</artifactId>
+      <version>${dep.version.flink}</version>
+      <scope>provided</scope>
+    </dependency>
+
+    <dependency>
+      <groupId>org.apache.hive</groupId>
+      <artifactId>hive-exec</artifactId>
+      <version>${dep.version.hive}</version>
+      <exclusions>
+        <exclusion>
+          <groupId>org.apache.orc</groupId>
+          <artifactId>orc-core</artifactId>
+        </exclusion>
+      </exclusions>
+    </dependency>
+
+  </dependencies>
+
+  <build>
+    <plugins>
+      <!-- The maven-shade plugin creates a fat jar that contains all
+        dependencies. -->
+      <plugin>
+        <groupId>org.apache.maven.plugins</groupId>
+        <artifactId>maven-shade-plugin</artifactId>
+        <version>3.4.1</version>
+        <configuration>
+          <shadedArtifactAttached>true</shadedArtifactAttached>
+          <shadedClassifierName>combined</shadedClassifierName>
+        </configuration>
+        <executions>
+          <execution>
+            <phase>package</phase>
+            <goals>
+              <goal>shade</goal>
+            </goals>
+          </execution>
+        </executions>
+      </plugin>
+      <plugin>
+        <groupId>org.apache.maven.plugins</groupId>
+        <artifactId>maven-compiler-plugin</artifactId>
+        <configuration>
+          <source>8</source>
+          <target>8</target>
+        </configuration>
+      </plugin>
+    </plugins>
+  </build>
+
+</project>
diff --git a/project-flink/flink-sql-parquet/src/main/java/com/github/lambda/lakehouse/FlinkAppKafkaToParquet.java b/project-flink/flink-sql-parquet/src/main/java/com/github/lambda/lakehouse/FlinkAppKafkaToParquet.java
new file mode 100644
index 0000000..50d4d7a
--- /dev/null
+++ b/project-flink/flink-sql-parquet/src/main/java/com/github/lambda/lakehouse/FlinkAppKafkaToParquet.java
@@ -0,0 +1,88 @@
+package com.github.lambda.lakehouse;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.Table;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class FlinkAppKafkaToParquet {
+
+  private static final Logger LOG = LoggerFactory.getLogger(FlinkAppKafkaToParquet.class);
+
+  public static void main(String[] args) throws Exception {
+
+    TableEnvironment tableEnv = buildTableEnvironment();
+
+    Table tableRawCustomers = buildSourceTable("raw_customers", tableEnv);
+    Table tableAggrCustomers = buildSinkTable("aggr_customers", tableEnv);
+
+    tableEnv.executeSql("INSERT INTO aggr_customers SELECT state, trace, worker_id FROM raw_customers");
+  }
+
+  public static Table buildSinkTable(String tableName, TableEnvironment tableEnv) {
+    String query = ""
+        + "CREATE TABLE " + tableName + " (\n"
+        + "   state STRING,\n"
+        + "   trace STRING,\n"
+        + "   worker_id STRING\n"
+        + ") PARTITIONED BY (worker_id) "
+        + "WITH (\n"
+        + " 'connector' = 'filesystem',\n"
+        + " 'format' = 'parquet',\n"
+        + " 'path' = 's3://datalake/parquet/test'\n"
+        + ");\n";
+    tableEnv.executeSql(query);
+    tableEnv.executeSql("SHOW CREATE TABLE " + tableName).print();
+
+    Table table = tableEnv.from(tableName);
+
+    return table;
+  }
+
+  public static Table buildSourceTable(String tableName, TableEnvironment tableEnv) {
+    String query = ""
+        + "CREATE TABLE " + tableName + " (\n"
+        + "   state STRING,\n"
+        + "   trace STRING,\n"
+        + "   worker_id STRING\n" + ") "
+        + "WITH (\n"
+        + "   'connector' = 'kafka',\n"
+        + "   'topic' = 'connect-cluster.inventory.status',\n"
+        + "   'properties.bootstrap.servers' = 'localhost:9092',\n"
+        + "   'properties.group.id' = 'testGroup',\n"
+        + "   'properties.auto.offset.reset' = 'earliest',\n"
+        + "   'scan.startup.mode' = 'earliest-offset',\n"
+        + "   'format' = 'json',"
+        + "   'json.ignore-parse-errors' = 'true',\n"
+        + "   'json.fail-on-missing-field' = 'false'\n"
+        + ");\n";
+    tableEnv.executeSql(query);
+    tableEnv.executeSql("SHOW CREATE TABLE " + tableName).print();
+
+    Table table = tableEnv.from(tableName);
+
+    return table;
+  }
+
+  public static StreamTableEnvironment buildTableEnvironment() {
+    // TODO (Kun): Handle Parameters
+    // - https://nightlies.apache.org/flink/flink-docs-master/docs/dev/datastream/application_parameters/
+
+    Configuration conf = new Configuration();
+
+    StreamExecutionEnvironment env = StreamExecutionEnvironment
+        .createLocalEnvironmentWithWebUI(conf);
+    env.getCheckpointConfig().setCheckpointInterval(10000L);
+//    env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
+//    env.setStateBackend(new EmbeddedRocksDBStateBackend());
+//    env.getCheckpointConfig().setCheckpointStorage("file:///tmp/flink-checkpoint");
+//    env.setDefaultSavepointDirectory("file:///tmp/flink-savepoint");
+
+    StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
+
+    return tableEnv;
+  }
+}
diff --git a/project-flink/flink-sql-parquet/src/main/resources/core-site.xml b/project-flink/flink-sql-parquet/src/main/resources/core-site.xml
new file mode 100644
index 0000000..b0dbd6e
--- /dev/null
+++ b/project-flink/flink-sql-parquet/src/main/resources/core-site.xml
@@ -0,0 +1,34 @@
+<configuration>
+  <property>
+    <name>fs.s3a.endpoint</name>
+    <value>http://localhost:9000</value>
+  </property>
+  <property>
+    <name>fs.s3a.access.key</name>
+    <value>minio</value>
+  </property>
+  <property>
+    <name>fs.s3a.secret.key</name>
+    <value>minio123</value>
+  </property>
+  <property>
+    <name>fs.s3a.path.style.access</name>
+    <value>true</value>
+  </property>
+  <property>
+    <name>fs.s3.endpoint</name>
+    <value>http://localhost:9000</value>
+  </property>
+  <property>
+    <name>fs.s3.access.key</name>
+    <value>minio</value>
+  </property>
+  <property>
+    <name>fs.s3.secret.key</name>
+    <value>minio123</value>
+  </property>
+  <property>
+    <name>fs.s3.path.style.access</name>
+    <value>true</value>
+  </property>
+</configuration>
diff --git a/project-flink/flink-sql-parquet/src/main/resources/log4j.properties b/project-flink/flink-sql-parquet/src/main/resources/log4j.properties
new file mode 100644
index 0000000..1ab559e
--- /dev/null
+++ b/project-flink/flink-sql-parquet/src/main/resources/log4j.properties
@@ -0,0 +1,5 @@
+log4j.rootLogger=INFO, console
+
+log4j.appender.console=org.apache.log4j.ConsoleAppender
+log4j.appender.console.layout=org.apache.log4j.PatternLayout
+log4j.appender.console.layout.ConversionPattern=%d{HH:mm:ss,SSS} %-5p %-60c %x - %m%n
diff --git a/project-flink/pom.xml b/project-flink/pom.xml
new file mode 100644
index 0000000..ec81124
--- /dev/null
+++ b/project-flink/pom.xml
@@ -0,0 +1,124 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+  xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+  xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+  <modelVersion>4.0.0</modelVersion>
+
+  <groupId>com.github.lambda.lakehouse</groupId>
+  <artifactId>project-flink</artifactId>
+  <version>1.0-SNAPSHOT</version>
+  <packaging>pom</packaging>
+
+  <modules>
+    <module>flink-sql-kafka</module>
+    <module>flink-sql-parquet</module>
+    <module>flink-sql-iceberg</module>
+  </modules>
+
+  <properties>
+    <java.version>8</java.version>
+    <maven.compiler.source>8</maven.compiler.source>
+    <maven.compiler.target>8</maven.compiler.target>
+
+    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+    <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
+
+    <dep.version.flink>1.16.1</dep.version.flink>
+    <dep.version.flinkShort>1.16</dep.version.flinkShort>
+    <dep.version.iceberg>1.3.1</dep.version.iceberg>
+    <dep.version.kafka>3.4.0</dep.version.kafka>
+    <dep.version.hadoop>3.3.1</dep.version.hadoop>
+    <dep.version.hive>3.1.2</dep.version.hive>
+    <dep.version.awssdk>2.20.18</dep.version.awssdk>
+
+    <dep.version.slf4j>1.7.36</dep.version.slf4j>
+    <dep.version.log4j>2.17.1</dep.version.log4j>
+
+    <checkstyle.violation.ignore>
+      NonEmptyAtclauseDescription,
+      JavadocMethod,
+      MissingJavadocType,
+      MissingJavadocMethod
+    </checkstyle.violation.ignore>
+  </properties>
+
+  <dependencies>
+    <dependency>
+      <groupId>org.apache.flink</groupId>
+      <artifactId>flink-core</artifactId>
+      <version>${dep.version.flink}</version>
+      <scope>provided</scope>
+    </dependency>
+
+    <dependency>
+      <groupId>org.apache.flink</groupId>
+      <artifactId>flink-java</artifactId>
+      <version>${dep.version.flink}</version>
+      <scope>provided</scope>
+    </dependency>
+
+    <!-- statebackend using rocksdb. -->
+    <dependency>
+      <groupId>org.apache.flink</groupId>
+      <artifactId>flink-statebackend-rocksdb</artifactId>
+      <version>${dep.version.flink}</version>
+    </dependency>
+
+    <!-- runtime-web dependency is need to start web UI from IDE -->
+    <dependency>
+      <groupId>org.apache.flink</groupId>
+      <artifactId>flink-runtime-web</artifactId>
+      <version>${dep.version.flink}</version>
+      <scope>provided</scope>
+    </dependency>
+
+    <!-- Add logging framework, to produce console output when running in the IDE. -->
+    <!-- These dependencies are excluded from the application JAR by default. -->
+    <dependency>
+      <groupId>org.slf4j</groupId>
+      <artifactId>slf4j-api</artifactId>
+      <version>${dep.version.slf4j}</version>
+      <scope>provided</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.logging.log4j</groupId>
+      <artifactId>log4j-slf4j-impl</artifactId>
+      <version>${dep.version.log4j}</version>
+      <scope>provided</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.logging.log4j</groupId>
+      <artifactId>log4j-api</artifactId>
+      <version>${dep.version.log4j}</version>
+      <scope>provided</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.logging.log4j</groupId>
+      <artifactId>log4j-core</artifactId>
+      <version>${dep.version.log4j}</version>
+      <scope>provided</scope>
+    </dependency>
+
+  </dependencies>
+
+  <dependencyManagement>
+    <dependencies>
+      <dependency>
+        <groupId>com.fasterxml.jackson</groupId>
+        <artifactId>jackson-bom</artifactId>
+        <type>pom</type>
+        <scope>import</scope>
+        <version>2.15.0</version>
+      </dependency>
+      <dependency>
+        <groupId>org.junit</groupId>
+        <artifactId>junit-bom</artifactId>
+        <version>5.9.3</version>
+        <type>pom</type>
+        <scope>import</scope>
+      </dependency>
+    </dependencies>
+  </dependencyManagement>
+
+
+</project>