Java Table API Quick Start on Confluent Cloud for Apache Flink

Confluent Cloud for Apache Flink®️ supports programming applications with the Table API. Confluent provides a plugin for running applications that use the Table API on Confluent Cloud.

For more information, see Table API.

For code examples, see Java Examples for Table API on Confluent Cloud.

Prerequisites

To run Table API and Flink SQL programs, you must generate an API key that’s specific to the Flink environment. Also, you need Confluent Cloud account details, like your organization and environment identifiers.

  • Flink API Key: Follow the steps in Generate a Flink API key. For convenience, assign your Flink key and secret to the FLINK_API_KEY and FLINK_API_SECRET environment variables.
  • Organization ID: The identifier your organization, for example, b0b421724-4586-4a07-b787-d0bb5aacbf87. For convenience, assign your organization identifier to the ORG_ID environment variable.
  • Environment ID: The identifier of the environment where your Flink SQL statements run, for example, env-z3y2x1. For convenience, assign your environment identifier to the ENV_ID environment variable.
  • Cloud provider name: The name of the cloud provider where your cluster runs, for example, aws. To see the available providers, run the confluent flink region list command. For convenience, assign your cloud provider to the CLOUD_PROVIDER environment variable.
  • Cloud region: The name of the region where your cluster runs, for example, us-east-1. To see the available regions, run the confluent flink region list command. For convenience, assign your cloud region to the CLOUD_REGION environment variable.
export CLOUD_PROVIDER="aws"
export CLOUD_REGION="us-east-1"
export FLINK_API_KEY="<your-flink-api-key>"
export FLINK_API_SECRET="<your-flink-api-secret>"
export ORG_ID="<your-organization-id>"
export ENV_ID="<your-environment-id>"
export COMPUTE_POOL_ID="<your-compute-pool-id>"

Compile and run a Table API program

The following code example shows how to run a “Hello World” statement and how to query an example data stream.

  1. Copy the following project object model (POM) into a file named pom.xml.

  2. Create a directory named “example”.

    mkdir example
    
  3. Create a file named hello_table_api.java in the example directory.

    touch example/hello_table_api.java
    
  4. Copy the following code into hello_table_api.java.

    package example;
    import io.confluent.flink.plugin.ConfluentSettings;
    import io.confluent.flink.plugin.ConfluentTools;
    import org.apache.flink.table.api.EnvironmentSettings;
    import org.apache.flink.table.api.Table;
    import org.apache.flink.table.api.TableEnvironment;
    import org.apache.flink.types.Row;
    import java.util.List;
    
    /**
     * A table program example to get started with the Apache Flink® Table API.
     *
     * <p>It executes two foreground statements in Confluent Cloud. The results of both statements are
     * printed to the console.
     */
    public class hello_table_api {
    
        // All logic is defined in a main() method. It can run both in an IDE or CI/CD system.
        public static void main(String[] args) {
    
            // Set up connection properties to Confluent Cloud.
            // Use the fromGlobalVariables() method if you assigned environment variables.
            // EnvironmentSettings settings = ConfluentSettings.fromGlobalVariables();
    
            // Use the fromArgs(args) method if you want to run with command-line arguments.
            EnvironmentSettings settings = ConfluentSettings.fromArgs(args);
    
            // Initialize the session context to get started.
            TableEnvironment env = TableEnvironment.create(settings);
    
            System.out.println("Running with printing...");
    
            // The Table API centers on 'Table' objects, which help in defining data pipelines
            // fluently. You can define pipelines fully programmatically.
            Table table = env.fromValues("Hello world!");
    
            // Also, You can define pipelines with embedded Flink SQL.
            // Table table = env.sqlQuery("SELECT 'Hello world!'");
    
            // Once the pipeline is defined, execute it on Confluent Cloud.
            // If no target table has been defined, results are streamed back and can be printed
            // locally. This can be useful for development and debugging.
            table.execute().print();
    
            System.out.println("Running with collecting...");
    
            // Results can be collected locally and accessed individually.
            // This can be useful for testing.
            Table moreHellos = env.fromValues("Hello Bob", "Hello Alice", "Hello Peter").as("greeting");
            List<Row> rows = ConfluentTools.collectChangelog(moreHellos, 10);
            rows.forEach(
                    r -> {
                        String column = r.getFieldAs("greeting");
                        System.out.println("Greeting: " + column);
                    });
        }
    }
    
  5. Run the following command to build the jar file.

    mvn clean package
    
  6. Run the jar. If you assigned your cloud configuration to the environment variables specified in the Prerequisites section, and you used the fromGlobalVariables method in the hello_table_api code, you don’t need to provide the command-line options.

    java -jar target/flink-table-api-java-hello-world-1.0.jar \
      --cloud aws \
      --region us-east-1 \
      --flink-api-key key \
      --flink-api-secret secret \
      --organization-id b0b21724-4586-4a07-b787-d0bb5aacbf87 \
      --environment-id env-z3y2x1 \
      --compute-pool-id lfcp-8m03rm
    

    Your output should resemble:

    Running with printing...
    +----+--------------------------------+
    | op |                             f0 |
    +----+--------------------------------+
    | +I |                   Hello world! |
    +----+--------------------------------+
    1 row in set
    Running with collecting...
    Greeting: Hello Bob
    Greeting: Hello Alice
    Greeting: Hello Peter