Apache Flink SubmitSql

Apache Flink SubmitSql

Certified

Execute SQL via Flink SQL Gateway

Runs a SQL statement through the Flink SQL Gateway without uploading a JAR. Respects session settings and waits until the operation reaches a terminal state.

yaml
type: io.kestra.plugin.flink.SubmitSql

Execute a streaming SQL query

yaml
id: flink-sql-streaming
namespace: company.team

tasks:
  - id: run-sql
    type: io.kestra.plugin.flink.SubmitSql
    gatewayUrl: "http://flink-sql-gateway:8083"
    statement: |
      INSERT INTO enriched_orders
      SELECT o.order_id, o.customer_id, c.name, o.amount, o.order_time
      FROM orders o
      JOIN customers c ON o.customer_id = c.id
    sessionConfig:
      catalog: "default_catalog"
      database: "default_database"
      configuration:
        execution.runtime-mode: "streaming"
        execution.checkpointing.interval: "30s"

Execute a batch SQL query

yaml
id: flink-sql-batch
namespace: company.team

tasks:
  - id: run-batch-sql
    type: io.kestra.plugin.flink.SubmitSql
    gatewayUrl: "http://flink-sql-gateway:8083"
    statement: |
      CREATE TABLE daily_summary AS
      SELECT DATE(order_time) as order_date,
             COUNT(*) as order_count,
             SUM(amount) as total_amount
      FROM orders
      WHERE order_time >= '2024-01-01'
      GROUP BY DATE(order_time)
    sessionConfig:
      configuration:
        execution.runtime-mode: "batch"
Properties

SQL Gateway URL

Base URL of the Flink SQL Gateway (e.g., http://flink-sql-gateway: 8083).

SQL statement

SQL statement to execute; supports DDL and DML.

SubTypestring

Acceptable terminal states

Operation states treated as success. For streaming include RUNNING to keep the session open. Defaults to ['FINISHED', 'RUNNING'].

Default30

Connection timeout

Connection timeout in seconds; defaults to 30.

Reference (ref) of the pluginDefaults to apply to this task.

Session configuration

Session options such as catalog, database, and configuration properties applied before execution.

Definitions
catalogstring
configurationobject
SubTypestring
databasestring

Session name

Optional session name; a temporary session is created when omitted.

Default300

Statement timeout

Execution timeout in seconds for the SQL statement; defaults to 300.

Operation handle

Unique identifier for the executed SQL operation.

Result count

Number of rows affected or returned; -1 when the gateway does not provide it.

Session handle

Identifier for the SQL Gateway session used.

Operation status

Final status reported by the SQL Gateway.

More Plugins in this Category

Liquibase
Liquibase

Tasks that run Liquibase change management commands.

Data

2 tasks

OPA (Open Policy Agent)
OPA (Open Policy Agent)
Enterprise

Tasks to manage Open Policy Agent policies and decisions, including upload, evaluation, and partial evaluation.

Infrastructure

5 tasks

3 blueprints