Search code examples
apache-flinkflink-streamingflink-sql

Apache Flink: How to query a relational database with the Table API?

The following code snippet is taken from this blog post:

val sensorTable = ??? // can be a CSV file, Kafka topic, database, or ...

// register the table source
tEnv.registerTableSource("sensors", sensorTable)

I would like to read data from a relational database. Does Flink have a TableSource for JDBC databases?

Solution

  • In its current version (1.4.0, Dec. 2017), Flink does not provide a built-in TableSource to ingest data from a relational database.

    However, there is a JDBCInputFormat that can be used. You can either

    • use it to read the data from the database using the DataSet API and register the DataSet as a Table or
    • implement a JdbcTableSource that wraps the JdbcInputFormat. A simple implementation of a JdbcTableSource should be easy to achieve. Implementing support for parallel reads, projection or filter push-down, requires more effort.