Flink Tour: Add Python examples

Author: amotl

framework/flink/tour/python/README.md

@@ -0,0 +1,28 @@
+# Apache Flink and CrateDB with Python
+
+Basic examples demonstrating how to read and write from/to
+CrateDB when using Apache Flink (PyFlink).
+
+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);'
+```
+Flink >= 1.19 has problems with JDBC and PyFlink,
+but previous versions need a Python of the same age.
+```shell
+uv venv --python 3.10 --seed .venv310
+uv pip install -r requirements.txt
+```
+```shell
+python write.py
+```
+```shell
+python ready.py
+```
+
+
+[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/python/read.py

@@ -0,0 +1,6 @@
+def main():
+    print("Hello World")
+
+
+if __name__ == "__main__":
+    main()

framework/flink/tour/python/requirements.txt

@@ -0,0 +1,2 @@
+apache-flink==1.18.1
+avro<1.13

framework/flink/tour/python/write.py

@@ -0,0 +1,80 @@
+import dataclasses
+import logging
+
+from pathlib import Path
+
+from pyflink.common import Types
+from pyflink.datastream import StreamExecutionEnvironment
+from pyflink.datastream.connectors.jdbc import JdbcSink, JdbcConnectionOptions, JdbcExecutionOptions
+
+
+logger = logging.getLogger(__name__)
+
+JARS_PATH = Path(__file__).parent / 'jars'
+
+
+@dataclasses.dataclass
+class Person:
+    name: str
+    age: int
+
+
+def main():
+
+    env = StreamExecutionEnvironment.get_execution_environment()
+    jars = list(map(lambda x: 'file://' + str(x), (JARS_PATH.glob('*.jar'))))
+    env.add_jars(*jars)
+
+    # Define source data.
+    ds = env.from_collection([
+        Person("Fred", 35),
+        Person("Wilma", 35),
+        Person("Pebbles", 2),
+    ])
+
+    # Define CrateDB as data sink.
+    row_type_info = Types.ROW_NAMED(['name', 'age'], [Types.STRING(), Types.INT()])
+    ds.add_sink(
+        JdbcSink.sink(
+            "INSERT INTO person (name, age) VALUES (?, ?)",
+            row_type_info,
+
+            # FIXME (Flink >= 1.19): java.lang.NoSuchMethodException: org.apache.flink.connector.jdbc.internal.JdbcOutputFormat.createRowJdbcStatementBuilder
+
+            # This is due to a bug in the python Flink library. In `flink-connector-jdbc` v3.1,
+            # the `JdbcOutputFormat` was renamed to `RowJdbcOutputFormat`. This change has up till
+            # now not been implemented in the python Flink library.
+            # https://stackoverflow.com/questions/78960829/java-lang-nosuchmethodexception-in-python-flink-jdbc
+
+            # As you see, java `JdbcSink` connector class has different shape from Python `JdbcSink` connector.
+            # In Java code, `jdbcSink` object is generated from `JdbcSinkBuilder` class, but in Python it is not.
+            # I think these errors are due to API version mismatch. Any idea to solve these errors?
+            # https://stackoverflow.com/questions/79604252/issue-with-pyflink-api-code-for-inserting-data-into-sql
+
+            JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
+            .with_url('jdbc:postgresql://localhost:5432/crate')
+            .with_driver_name('org.postgresql.Driver')
+            .with_user_name("crate")
+            .with_password("crate")
+            .build(),
+            JdbcExecutionOptions.builder()
+            .with_batch_interval_ms(1000)
+            .with_batch_size(200)
+            .with_max_retries(5)
+            .build()
+        )
+    )
+
+    # Execute pipeline.
+    env.execute()
+
+
+if __name__ == '__main__':
+    logging.basicConfig(
+        format='[%(asctime)s] [%(levelname)s] [%(filename)s:%(lineno)d] - %(message)s',
+        level=logging.DEBUG
+    )
+
+    logger.info("Start")
+    main()
+    logger.info("Ready")