Flink Tour: Add Java examples

Author: amotl

.gitignore

@@ -4,9 +4,11 @@ __pycache__
 _build
 .coverage*
 .env
+.gradle
 .DS_Store
 .meltano
 .pytest_cache
+build
 coverage.xml
 mlruns/
 archive/

framework/flink/tour/java/README.md

@@ -0,0 +1,22 @@
+# Apache Flink and CrateDB with Java
+
+Basic examples demonstrating how to read and write from/to
+CrateDB when using Apache Flink.
+
+The examples use both the [CrateDB JDBC] and the [PostgreSQL JDBC]
+driver. CrateDB JDBC is needed for catalog operations, which are
+required when reading from CrateDB using Flink.
+
+```sql
+uvx crash -c 'CREATE TABLE person (name STRING, age INT);'
+```
+```shell
+gradle run write
+```
+```shell
+gradle run read
+```
+
+
+[CrateDB JDBC]: https://cratedb.com/docs/guide/connect/java/cratedb-jdbc.html
+[PostgreSQL JDBC]: https://cratedb.com/docs/guide/connect/java/postgresql-jdbc.html

framework/flink/tour/java/build.gradle

@@ -0,0 +1,16 @@
+// https://medium.com/@santoshkotagiri/setting-up-flink-projects-with-gradle-a-step-by-step-guide-4d1ef85f017
+apply plugin : 'application'
+repositories.mavenCentral()
+dependencies {
+    implementation 'org.apache.flink:flink-clients:1.20.3'
+    implementation 'org.apache.flink:flink-connector-jdbc:3.2.0-1.19'
+    implementation 'org.apache.flink:flink-table-planner_2.12:1.20.3'
+    implementation 'org.apache.flink:flink-table-runtime:1.20.3'
+    implementation 'org.slf4j:slf4j-simple:1.7.36'
+    runtimeOnly    'io.crate:crate-jdbc:2.7.0'
+    runtimeOnly    'org.postgresql:postgresql:42.7.8'
+}
+java.toolchain.languageVersion = JavaLanguageVersion.of(11)
+sourceSets.main.java.srcDirs = ['.']
+tasks.register('read') { application.mainClass = 'read' }
+tasks.register('write') { application.mainClass = 'write' }

framework/flink/tour/java/read.java

@@ -0,0 +1,51 @@
+// Read from CrateDB using Apache Flink.
+// Invoke: gradle run read
+// Source: https://github.com/crate/cratedb-examples/blob/main/framework/flink/tour/java/read.java
+
+// https://tacnode.io/docs/guides/ecosystem/bigdata/flink#catalog-registration
+// https://github.com/crate/cratedb-flink-jobs/blob/main/src/main/java/io/crate/flink/demo/SimpleTableApiJob.java
+// https://nightlies.apache.org/flink/flink-docs-release-1.20/docs/connectors/table/jdbc/#jdbc-catalog-for-postgresql
+import org.apache.flink.connector.jdbc.catalog.JdbcCatalog;
+import org.apache.flink.table.api.*;
+
+import static org.apache.flink.table.api.Expressions.$;
+
+public class read {
+
+    public static String CATALOG_NAME = "example_catalog";
+
+    public static void main(String[] args) throws Exception {
+
+        // Create Flink Table API environment.
+        EnvironmentSettings settings = EnvironmentSettings.inStreamingMode();
+        TableEnvironment env = TableEnvironment.create(settings);
+
+        // Define catalog.
+        // CrateDB only knows a single database called `crate`,
+        // but you can separate concerns using schemata. The
+        // default schema is `doc`.
+        JdbcCatalog catalog = new JdbcCatalog(
+            CATALOG_NAME,
+            "crate",
+            "crate",
+            "crate",
+            "jdbc:crate://localhost:5432"
+            );
+
+        // Register catalog and set as default.
+        env.registerCatalog(CATALOG_NAME, catalog);
+        env.useCatalog(CATALOG_NAME);
+
+        // Invoke query using plain SQL.
+        // FIXME: Currently does not work with `sys.summits`.
+        // SqlValidatorException: Object 'sys.summits' not found
+        env.executeSql("SELECT * FROM `doc.person` LIMIT 3").print();
+
+        // Invoke query using DSL.
+        env.from("`doc.person`")
+                .select($("name"), $("age"))
+                .execute()
+                .print();
+    }
+
+}

framework/flink/tour/java/write.java

@@ -0,0 +1,57 @@
+// Write to CrateDB using Apache Flink.
+// Invoke: uvx crash -c 'CREATE TABLE person (name STRING, age INT);'
+// Invoke: gradle run write
+// Source: https://github.com/crate/cratedb-examples/blob/main/framework/flink/tour/java/write.java
+
+// https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/jdbc/#full-example
+
+import org.apache.flink.connector.jdbc.*;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+
+public class write {
+
+    public static void main(String[] args) throws Exception {
+        var env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+        // Define source data.
+        env.fromElements(
+            new Person("Fred", 35),
+            new Person("Wilma", 35),
+            new Person("Pebbles", 2)
+
+        // Define CrateDB as data sink.
+        ).addSink(
+            JdbcSink.sink(
+                "INSERT INTO person (name, age) VALUES (?, ?)",
+                (statement, person) -> {
+                    statement.setString(1, person.name);
+                    statement.setInt(2, person.age);
+                },
+                JdbcExecutionOptions.builder()
+                    .withBatchSize(1000)
+                    .withBatchIntervalMs(200)
+                    .withMaxRetries(5)
+                    .build(),
+                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
+                    .withUrl("jdbc:postgresql://localhost:5432/doc?sslmode=disable")
+                    .withDriverName("org.postgresql.Driver")
+                    .withUsername("crate")
+                    .withPassword("crate")
+                    .build()
+            ));
+
+        // Execute pipeline.
+        env.execute();
+    }
+
+    public static class Person {
+        public String name;
+        public Integer age;
+        public Person() {}
+        public Person(String name, Integer age) {
+            this.name = name;
+            this.age = age;
+        }
+    }
+
+}