[refactor] update iceberg poc

Author: feliux

docker/iceberg/README.md

@@ -2,6 +2,8 @@
 
 ```sh
 $ git clone git@github.com:tabular-io/docker-spark-iceberg.git
+$ cd docker-spark-iceberg
+$ docker-compose up -d
 $ docker exec -it spark-iceberg pyspark
 ```
 

docker/iceberg/iceberg.py docker/iceberg/kinesis-table.pydocker/iceberg/iceberg.py renamed to docker/iceberg/kinesis-table.py

@@ -0,0 +1,50 @@
+# Creating table
+# %%sql
+# CREATE DATABASE IF NOT EXISTS climate;
+spark.sql("CREATE DATABASE IF NOT EXISTS climate")
+
+# %%sql
+# CREATE TABLE IF NOT EXISTS climate.weather (
+#     datetime              timestamp,
+#     temp                  double,
+#     lat                   double,
+#     long                  double,
+#     cloud_coverage        string,
+#     precip                double,
+#     wind_speed            double
+# )
+# USING iceberg
+# PARTITIONED BY (days(datetime))
+spark.sql("CREATE TABLE IF NOT EXISTS climate.weather ( \
+    datetime              timestamp, \
+    temp                  double, \
+    lat                   double, \
+    long                  double, \
+    cloud_coverage        string, \
+    precip                double, \
+    wind_speed            double \
+) \
+USING iceberg \
+PARTITIONED BY (days(datetime))")
+
+# Writing data
+from datetime import datetime
+schema = spark.table("climate.weather").schema
+data = [
+    (datetime(2023,8,16), 76.2, 40.951908, -74.075272, "Partially sunny", 0.0, 3.5),
+    (datetime(2023,8,17), 82.5, 40.951908, -74.075272, "Sunny", 0.0, 1.2),
+    (datetime(2023,8,18), 70.9, 40.951908, -74.075272, "Cloudy", .5, 5.2)
+  ]
+df = spark.createDataFrame(data, schema)
+df.writeTo("climate.weather").append()
+
+
+# Reading data
+from pyiceberg.catalog import load_catalog
+from pyiceberg.expressions import GreaterThanOrEqual
+
+catalog = load_catalog("default")
+tbl = catalog.load_table("climate.weather")
+
+sc = tbl.scan(row_filter=GreaterThanOrEqual("datetime", "2023-08-01T00:00:00.000000+00:00"))
+df = sc.to_arrow().to_pandas()

docker/iceberg/weather-table.py

@@ -10,8 +10,10 @@ import (
 )
 
 const (
-	connStr string = "postgres://root@localhost:4567/dev"
-	sqlFile string = "./sql/kinesis-to-iceberg.sql"
+	connStr       string = "postgres://root@localhost:4567/dev"
+	sqlFile       string = "./sql/kinesis-to-iceberg.sql"
+	sourceJobName string = "create-kinesis-source"
+	sinkJobName   string = "create-iceberg-sink"
 )
 
 func main() {
@@ -26,13 +28,13 @@ func main() {
 		log.Fatalf("Error loading sql: %v\n", err)
 	}
 	// Source
-	_, err = dot.Exec(db, "create-kinesis-source")
+	_, err = dot.Exec(db, sourceJobName)
 	if err != nil {
 		log.Fatalf("Error executing create table: %v\n", err)
 	}
 
 	// Sink
-	_, err = dot.Exec(db, "create-iceberg-sink")
+	_, err = dot.Exec(db, sinkJobName)
 	if err != nil {
 		log.Fatalf("Error executing create sink: %v\n", err)
 	}