Key Search with Confluent Cloud for Apache Flink

Confluent Cloud for Apache Flink® supports read-only external tables to enable key search with federated query execution on external data source, like JDBC, REST, and MongoDB.

Note

Key Search is an Open Preview feature in Confluent Cloud.

A Preview feature is a Confluent Cloud component that is being introduced to gain early feedback from developers. Preview features can be used for evaluation and non-production testing purposes or to provide feedback to Confluent. The warranty, SLA, and Support Services provisions of your agreement with Confluent do not apply to Preview features. Confluent may discontinue providing preview releases of the Preview features at any time in Confluent’s’ sole discretion.

Use the KEY_SEARCH_AGG function to run key searches over external databases in Confluent Cloud for Apache Flink.

SELECT * FROM key_input,
  LATERAL TABLE(KEY_SEARCH_AGG(<external_table>, DESCRIPTOR(<input_column>), <search_column>))

Supported providers

These key search providers are supported:

• Confluent-jdbc: Postgres, MySQL, SQL Server, Oracle
• REST
• MongoDB
• Couchbase

Examples

Search table parameters

Common parameters

{CONNECTOR}

Provider name.

{CONNECTOR}.PARAMS.*

Parameters supported by external providers.

Confluent-jdbc properties

Confluent-jdbc.connection

Confluent JDBC connection that includes the endpoint, username, and password.

This property is required.

Confluent-jdbc.table-name

The name of JDBC table to connect.

This property is required.

Rest properties

Rest.connection

REST connection that includes the endpoint and credentials required to connect to the endpoint.

The REST endpoint must use the https protocol, can’t be an ip address, and can’t end with confluent.cloud as the domain name.

This property is required.

Rest.path

REST path segments under the base endpoint in the connection for issuing HTTP requests toward the endpoint.

This property is required.

Rest.method

HTTP verb that will be used to invoke the RESTful endpoint

This property is required.

Mongodb properties

Mongodb.connection

MongoDB connection that includes the endpoint, username, and passwords.

This property is required.

Mongodb.database

MongoDB database name.

This property is required.

Mongodb.collection

MongoDB collection name.

This property is required.

Mongodb.index

MongoDB search index name.

This property is required.

Examples

Key search with Postgres

This example shows how to run key search with Postgres by using Flink SQL.

This example shows the following steps:

• Create a connection resource to a Postgres database.
• Create a Confluent-jdbc external table.
• Create an input text table.
• Run a key search on movie titles.
• Show the results and store them in a new table.

1. In the Flink SQL shell, or in a Flink workspace in Confluent Cloud Console, run the following statement to create a connection resource named jdbc-postgres-connection that uses your Postgres credentials.

   CREATE CONNECTION jdbc-postgres-connection
     WITH (
       'type' = 'confluent-jdbc',
       'endpoint' = '<jdbc-url>',
       'username' = '<username>',
       'password' = '<password>'
     );
   

2. Create a Postgres table with preset example data like Netflix shows. Confluent-jdbc also supports mysql, sql-server, and oracle.

   -- create jdbc table
   CREATE TABLE jdbc_postgres (
       show_id STRING,
       type STRING,
       title STRING,
       cast_members STRING,
       country STRING,
       date_added DATE,
       release_year INT,
       rating STRING,
       duration STRING,
       listed_in STRING,
       description STRING
   ) WITH (
       'connector' = 'confluent-jdbc',
       'confluent-jdbc.connection' = 'jdbc-postgres-connection',
       'confluent-jdbc.table-name' = 'netflix_shows'
   );
   

3. Run the following statements to create an input table and insert data.

   -- create input table
   CREATE TABLE jdbc_postgres_input_table (
       title STRING
   );
   
   INSERT INTO jdbc_postgres_input_table VALUES
     ('Squid Game'),
     ('The Irishman'),
     ('Stranger Things');
   

4. Run the following statement to run key search on movie title.

   ---- key search
   SELECT * FROM jdbc_postgres_input_table,
     LATERAL TABLE(KEY_SEARCH_AGG(jdbc_postgres, DESCRIPTOR(title), title));
   

5. Run the following statement to store search results in a new table.

   -- store search results in a new table
   CREATE TABLE jdbc_postgres_result_table AS
     SELECT * FROM jdbc_postgres_input_table,
       LATERAL TABLE(KEY_SEARCH_AGG(jdbc_postgres, DESCRIPTOR(title), title));
   
   SELECT * FROM jdbc_postgres_result_table
     CROSS JOIN UNNEST(search_results)
     AS T(id, text_column, integer_column, float_column, double_column, numeric_column, boolean_column, timestamp_column)
   

Key search with REST

This example shows how to run key search with any RESTful server by using Flink SQL. It assumes the RESTful server’s response format is in JSON.

This example shows the following steps:

• Create a connection resource to a RESTful server.
• Create a REST external table.
• Create a text input table.
• Run a key search on book titles.
• Show the results and store them in a new table.

1. In the Flink SQL shell, or in a Flink workspace in Confluent Cloud Console, run the following statement to create a connection resource named rest-connection that encapsulates the endpoint and the credentials that will be used to access the RESTful server.

   CREATE CONNECTION rest-connection
     WITH (
       'type' = 'rest',
       'endpoint' = '<endpoint>',
       'username' = '<username>',
       'password' = '<password>'
     );
   

2. Run the following statement to create a REST external table.

   -- REST key search
   CREATE TABLE rest_books_key_search (
     numFound BIGINT,
     q STRING NOT NULL,
     documentation_url STRING,
     docs ARRAY<ROW<`author_name` ARRAY<STRING>, `title` STRING>>,
     PRIMARY KEY (q) NOT ENFORCED
   ) WITH (
     'connector' = 'rest',
     'rest.connection' = 'rest-connection-openlib',
     'rest.method' = 'GET',
     'rest.path' = 'search.json'
   );
   

3. Run the following statement to create a text input table.

   -- create search table
   CREATE TABLE bookstore (
     bookname STRING NOT NULL,
     price INT,
     stock INT
   );
   

4. Run the following statement to insert data into the table.

   INSERT INTO `bookstore` VALUES
     ('harry-potter', 20, 500),
     ('wish', 10, 1000),
     ('crucial conversation', 10, 2000)
   

5. Run the following statement to run key search on book titles.

   -- REST key search
   SELECT * FROM bookstore,
     LATERAL TABLE (KEY_SEARCH_AGG(rest_books_key_search, DESCRIPTOR(q), bookname))
   

6. Run the following statement to store search results in a new table.

   -- store search results in a new table
   CREATE TABLE books_key_search_results AS
     SELECT * FROM bookstore,
       LATERAL TABLE (KEY_SEARCH_AGG(rest_books_key_search, DESCRIPTOR(q), bookname))
   

7. Run the following statement to show the results.

   SELECT * FROM books_key_search_results
     CROSS JOIN UNNEST(search_results)
     AS T(numFound, q, documentation_url, docs)
   

Key search with MongoDB

This example shows how to run a key search with MongoDB by using Flink SQL.

This example shows the following steps:

• Create a connection resource to a MongoDB database.
• Create a MongoDB external table.
• Create a text input table.
• Run a key search on movie titles.
• Show the results and store them in a new table.

It assumes an Atlas cluster as shown in Create a Cluster and MongoDB sample dataset.

1. In the Flink SQL shell, or in a Flink workspace in Confluent Cloud Console, run the following statement to create a connection resource named mongodb-connection that uses your MongoDB credentials.

   CREATE CONNECTION mongodb-connection
     WITH (
       'type' = 'mongodb',
       'endpoint' = '<atlas_endpoint>',
       'username' = '<atlas_username>',
       'password' = '<atlas_password>'
     );
   

2. Run the following statement to create a MongoDB external table.

   -- mongodb key search
   CREATE TABLE mongodb_movies_key_search (
     title STRING,
     plot STRING
   ) WITH (
     'connector' = 'mongodb',
     'mongodb.connection' = 'mongodb-connection',
     'mongodb.database' = 'sample_mflix',
     'mongodb.collection' = 'movies',
     'mongodb.index' = 'default'
   );
   

3. Run the following statement to create a search table.

   -- create search table
   CREATE TABLE movies_key_search(
     title STRING,
     year INT
   );
   

4. Run the following statement to insert data into the search table.

   INSERT INTO movies_key_search VALUES
     ('golden finger', 1964),
     ('license to kill', 1989),
     ('skyfall', 2012),
     ('die hard', 1988)
   

5. Run the following statement to run a key search on movie titles.

   -- mongodb key search
   SELECT * FROM movies_key_search,
     LATERAL TABLE(KEY_SEARCH_AGG(`mongodb_movies_key_search`, descriptor(title), title))
   

6. Run the following statement to store search results in a new table.

   -- store search results in a new table
   CREATE TABLE movies_key_search_results AS SELECT * FROM movies_key_search,
     LATERAL TABLE(KEY_SEARCH_AGG(`mongodb_movies_key_search`, descriptor(title), title))
   

7. Run the following statement to show the results.

   SELECT * FROM movies_key_search_results
     CROSS JOIN UNNEST(search_results)
     AS T(title, plot)
   

Related content

• KEY_SEARCH_AGG function
• CREATE MODEL statement
• AI Model Inference Functions

Note

This website includes content developed at the Apache Software Foundation under the terms of the Apache License v2.